python类Resource()的实例源码

main.py 文件源码 项目:syntaxnet-rest-api 作者: ljm625 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def post(self):
        args = self.reqparse.parse_args()
        if parse_handler==None:
            return {"result":"fail", "reason":"Please initialize the model first!"}, 400
        #parse = parser.SyntaxnetParser(segmenter_model,parser_model,folder=args['syntax_folder'])
        try:
            return parse_handler.parse_multi_string(args['strings'],tree=args['tree'])
        except Exception as e:
            return {'result': 'fail', "reason": str(e)}, 400

# class SyntaxModelQuery(Resource):
#     def __init__(self):
#         self.reqparse = reqparse.RequestParser()
#         self.reqparse.add_argument('strings', required=True, type=list, help='string is required field, you should input a list of strings', location='json')
#         self.reqparse.add_argument('syntax_folder', required=False, type=str, default=config.syntaxnetFolder,
#                                    location='json')
#         super(SyntaxModelQuery, self).__init__()
#
#     def post(self,folder):
#         args = self.reqparse.parse_args()
#         parse = parser.SyntaxnetParser(folder=args['syntax_folder'])
#         try:
#             return parse.parse_multi_string_custom(args['strings'],folder=folder)
#         except Exception, e:
#             return {'result': 'fail', "reason": e}, 400
utils.py 文件源码 项目:osp-api 作者: opensyllabus 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def add_cache_timestamp(resource_method):
        """Resource method decorator which adds a
        timestamp to the JSON response.

        Since responses are cached the timestamp will
        effectively indicate the time/version of the cached
        response.

        """

        @wraps(resource_method)
        def wrapped_resource_method(*args, **kwargs):
            cache_timestamp = {
                'cacheTimestamp': datetime.utcnow().isoformat() + 'Z',
            }
            response = resource_method(*args, **kwargs)
            try:
                response['meta'].update(cache_timestamp)
            except KeyError:
                response['meta'] = cache_timestamp
            return response

        return wrapped_resource_method
api.py 文件源码 项目:SuperFlaskSeed 作者: saurabh1e 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def get(self, slug=None):
        if slug:
            obj = self.resource.model.query.filter(self.resource.model.id == slug)
            obj = self.resource.has_read_permission(obj).first()
            if obj:
                return make_response(jsonify(self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                  only=tuple(self.resource.obj_only)).dump(
                    obj, many=False).data), 200)

            return make_response(jsonify({'error': True, 'message': 'Resource not found'}), 404)

        else:
            objects = self.resource.apply_filters(queryset=self.resource.model.query, **request.args)
            objects = self.resource.has_read_permission(objects)

            if '__order_by' in request.args:
                objects = self.resource.apply_ordering(objects, request.args['__order_by'])
            resources = objects.paginate(page=self.resource.page, per_page=self.resource.limit)
            if resources.items:
                return make_response(jsonify({'success': True,
                                              'data': self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                           only=tuple(self.resource.obj_only))
                                             .dump(resources.items, many=True).data, 'total': resources.total}), 200)
            return make_response(jsonify({'error': True, 'message': 'No Resource Found'}), 404)
api.py 文件源码 项目:open-pos-api 作者: saurabh1e 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def get(self, slug=None):
        if slug:
            obj = self.resource.model.query.filter(self.resource.model.id == slug)
            obj = self.resource.has_read_permission(obj).first()
            if obj:
                return make_response(jsonify(self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                  only=tuple(self.resource.obj_only)).dump(
                    obj, many=False).data), 200)

            return make_response(jsonify({'error': True, 'message': 'Resource not found'}), 404)

        else:
            objects = self.resource.apply_filters(queryset=self.resource.model.query, **request.args)
            objects = self.resource.has_read_permission(objects)

            if '__order_by' in request.args:
                objects = self.resource.apply_ordering(objects, request.args['__order_by'])
            resources = objects.paginate(page=self.resource.page, per_page=self.resource.limit)
            if resources.items:
                return make_response(jsonify({'success': True,
                                              'data': self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                           only=tuple(self.resource.obj_only))
                                             .dump(resources.items, many=True).data, 'total': resources.total}), 200)
            return make_response(jsonify({'error': True, 'message': 'No Resource Found'}), 404)
server.py 文件源码 项目:Chorus 作者: DonaldBough 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def auth2Token(code):
    url = "https://accounts.spotify.com/api/token"
    grant_type = "authorization_code"
    #get code from UI
    redirect_uri = "http://localhost:8000/create.html"
    #redirect_uri = "http:%2F%2Flocalhost:8000%2Fcreate.html"
    client_id = "0abc049d139f4ee8b465fd9416aa358d"
    client_secret = "dd477b353e744461ae1b3062f256c952"
    payload = {'grant_type': grant_type, 'code': code, 'redirect_uri': redirect_uri, 'client_id':client_id, 'client_secret':client_secret}

    req = requests.post(url, data = payload)
    res = json.loads (req.content)

    return [res['access_token'], res['refresh_token']]

#class CreateUser(Resource):
server.py 文件源码 项目:Chorus 作者: DonaldBough 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def CreateUser(currentEvent, inEvent, isHost):
    try:
        # Parse the arguments
        '''
        parser = reqparse.RequestParser()
        parser.add_argument('currentEvent', type=str)
        parser.add_argument('inEvent', type=str)
        parser.add_argument('host', type=str)
        args = parser.parse_args()
        '''     


        db = Database()
        userID = db.insertUser(currentEvent, inEvent, isHost)

        return userID

    except Exception as e:
        return str(e)

#class CreateHost(Resource):
emulator.py 文件源码 项目:Redfish-Interface-Emulator 作者: DMTF 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def init_resource_manager():
    """
    Initializes the resource manager
    """
    global resource_manager
    global REST_BASE
    global TRAYS
    global SPEC
    resource_manager = ResourceManager(REST_BASE, SPEC,MODE,TRAYS)

    # If POPULATE is specified in emulator-config.json, INFRAGEN is called to populate emulator (i.e. with Chassi, CS, Resource Blocks, etc) according to specified file
    try:
        POPULATE
    except:
        pass
    else:
        if os.path.exists(POPULATE):
            with open(POPULATE, 'r') as f:
                infragen_config = json.load(f)
            populate(infragen_config.get('POPULATE',10))

    resource_dictionary = ResourceDictionary()
emulator.py 文件源码 项目:Redfish-Interface-Emulator 作者: DMTF 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def get_configuration(obj, path):
        """
        Helper function to follow the given path starting with the given object

        Arguments:
            obj  - Beginning object to start searching down.  obj should have a get_resource()
            path - Path of object to get

        """

        try:
            config = obj.get_resource(path)
        except (IndexError, AttributeError, TypeError, AssertionError, KeyError) as e:
            traceback.print_exc()
            raise PathError("Resource not found: " + str(e.message))
        print (config)
        return config

#
# If DELETE /redfish/v1/reset, then reset the resource manager
#
ComputerSystem_api.py 文件源码 项目:Redfish-Interface-Emulator 作者: DMTF 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def post(self):
        resp = INTERNAL_ERROR
        req = request.get_json()

        if req is not None:
            composed_system = CreateComposedSystem(req)
            resp = composed_system, 200
        else:
            resp = INTERNAL_ERROR

        return resp


#class ComposedSystem(Resource):
#    def __init__(self):
#        pass
emulator_ssl.py 文件源码 项目:Redfish-Interface-Emulator 作者: DMTF 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def get_configuration(obj, path):
        """
        Helper function to follow the given path starting with the given object

        Arguments:
            obj  - Beginning object to start searching down.  obj should have a get_resource()
            path - Path of object to get

        """

        try:
            config = obj.get_resource(path)
        except (IndexError, AttributeError, TypeError, AssertionError, KeyError) as e:
            traceback.print_exc()
            raise PathError("Resource not found: " + str(e.message))
        return config

#
# If DELETE /redfish/v1/reset, then reset the resource manager
#
charity.py 文件源码 项目:donatemates 作者: donatemates 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def __init__(self):
        super(Resource, self).__init__()
stats.py 文件源码 项目:donatemates 作者: donatemates 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def __init__(self):
        super(Resource, self).__init__()
        self.campaign_table = DynamoTable('campaigns')
        self.donation_table = DynamoTable('donations')
campaign.py 文件源码 项目:donatemates 作者: donatemates 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def __init__(self):
        super(Resource, self).__init__()
        self.table = DynamoTable('campaigns')
campaign.py 文件源码 项目:donatemates 作者: donatemates 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def __init__(self):
        super(Resource, self).__init__()
        self.campaign_table = DynamoTable('campaigns')
        self.donation_table = DynamoTable('donations')
campaign.py 文件源码 项目:donatemates 作者: donatemates 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def __init__(self):
        super(Resource, self).__init__()
        self.campaign_table = DynamoTable('campaigns')
volume.py 文件源码 项目:cobalt 作者: PressLabs 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def put(volume_id):
        """Edits a volume dict based on the given representation.

        Expects to receive a json payload with a complete new version of the volume to be edited

        Args:
            volume_id (str): The id parsed from the URL

        Returns:
             tuple: payload, http status code, headers
        """
        manager = app.volume_manager

        volume = manager.by_id(volume_id)
        if volume is None:
            return {'message': 'Not Found'}, 404

        if volume.value['state'] != 'ready':
            return {'message': 'Resource not in ready state, can\'t update.'}, 409

        new_volume, errors = VolumeAttributeSchema().load(request.get_json(force=True))
        if errors:
            return {'message': errors}, 400

        if volume.value['requested'] == new_volume:
            return '', 304

        volume.value['requested'] = new_volume
        volume.value['state'] = 'pending'

        volume = manager.update(volume)
        if not volume:
            return {'message': 'Resource changed during transition.'}, 409

        result, _ = VolumeSchema().dump(volume)
        return result, 202, {'Location': app.api.url_for(Volume, volume_id=result['id'])}
volume.py 文件源码 项目:cobalt 作者: PressLabs 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def delete(volume_id):
        """Deletes the volume pointed by the id.

        Args:
            volume_id (str): The id parsed from the URL

        Returns:
            tuple: payload, http status code, headers
        """
        manager = app.volume_manager

        target_volume = manager.by_id(volume_id)
        if target_volume is None:
            return {'message': 'Not Found'}, 404

        if target_volume.value['state'] != 'ready':
            return {'message': 'Resource not in ready state, can\'t delete.'}, 409

        lock = manager.get_lock(volume_id, 'clone')
        lock.acquire(timeout=0, lock_ttl=10)

        pending_clones = []
        for volume in manager.all()[1]:
            if volume.value['control']['parent_id'] == volume_id:
                pending_clones.append(manager.get_id_from_key(volume.key))

        if pending_clones:
            lock.release()
            return {'message': 'Resource has pending clones, can\'t delete.',
                    'clones': pending_clones}, 409

        target_volume.value['state'] = 'deleting'
        target_volume = manager.update(target_volume)
        lock.release()

        if not target_volume:
            return {'message': 'Resource changed during transition.'}, 409

        result, _ = VolumeSchema().dump(target_volume)
        return result, 202, {'Location': app.api.url_for(Volume, volume_id=result['id'])}
flask.py 文件源码 项目:doctor 作者: upsight 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def __init__(self, schema_dir, resource_schema_class=None,
                 default_base_handler=None,
                 raise_response_validation_errors=False):
        if resource_schema_class is None:
            resource_schema_class = FlaskResourceSchema
        if default_base_handler is None:
            default_base_handler = flask_restful.Resource
        super(FlaskRouter, self).__init__(
            schema_dir, resource_schema_class, default_base_handler,
            raise_response_validation_errors)
utils.py 文件源码 项目:osp-api 作者: opensyllabus 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def resource_method_info(self, method) -> tuple:
        """Return the (class) name of the resource and its type, which
        is associated with the provided method (function).

        Used by this class' decorators to determine the name of the
        instance which subclassed this class, e.g., `WorksCollection`,
        as well as passing a nice `enum` for determining if it's
        a collection or singleton resource.

        Arguments:
            method (function): Hopefully a method which belongs
                to an instance that inherited from this class.

        Raises:
            InvalidResourceName: If the function belongs to
                a resource class whose name neither ends in
                `Collection` nor `Singleton`.

        Returns:
            tuple[str, str]: Resource class name, resoure type.

        """

        resource_class_name = self.__class__.__name__
        if resource_class_name.endswith('Collection'):
            resource_type = ResourceType.collection
        elif resource_class_name.endswith('Singleton'):
            resource_type = ResourceType.singleton
        else:
            raise InvalidResourceName(resource_class_name)

        return resource_class_name, resource_type
utils.py 文件源码 项目:osp-api 作者: opensyllabus 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def add_year_restriction(self, function):
        """Resource method decorator which applies a restriction
        only allowing results related to a syllabus of one year of
        age or older.

        This only works for HTTP GET methods on collection resources.
        Other methods will be unaffected entirely by this decorator.

        """

        method = function.__name__
        __, resource_type = self.resource_method_info(function)
        if method == "get" and resource_type == ResourceType.collection:
            @wraps(function)
            def new_function_to_replace_old(*args, **kwargs):
                request_args = {
                    'useOldIndex': True
                }
                request_args.update(flask.request.args)
                flask.request.args = ImmutableMultiDict(request_args)
                return function(*args, **kwargs)
            return new_function_to_replace_old
        else:
            return function


# TODO, FIXME: bad, remove this ASAP
# (this is here as filler because we haven't gotten
# to the elasticsearch bit yet)
test_everything.py 文件源码 项目:osp-api 作者: opensyllabus 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def test_custom_role_limits(self):

        def before_create_app():
            """Create a resource which uses overrides for
            role limits.

            """

            class TestResource(Resource):
                decorators = [
                    limits.limiter.limit(
                        limits.RoleBasedLimits(free='1 per day'),
                    ),
                ]

                def get(self):
                    return {'lol': 'lol!'}

            api.add_resource(TestResource, '/test')

        client = setup_client(before_create_app)

        response = client.get('/v1/test')
        data = json.loads(response.get_data(as_text=True))
        assert 'lol' in data

        response = client.get('/v1/test')
        assert response.status_code == 429
api.py 文件源码 项目:SuperFlaskSeed 作者: saurabh1e 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def get(self, slug=None):
        if slug:
            obj = self.resource.model.query.filter(self.resource.model.id == slug)
            obj = self.resource.has_read_permission(obj).first()
            if obj:
                return make_response(jsonify(self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                  only=tuple(self.resource.obj_only)).dump(
                    obj, many=False).data), 200)

            return make_response(jsonify({'error': True, 'message': 'Resource not found'}), 404)

        else:
            objects = self.resource.apply_filters(queryset=self.resource.model.query, **request.args)
            objects = self.resource.has_read_permission(objects)

            if '__order_by' in request.args:
                objects = self.resource.apply_ordering(objects, request.args['__order_by'])

            if '__export__' in request.args and self.resource.export is True:
                objects = objects.paginate(page=self.resource.page, per_page=self.resource.max_export_limit)
                return make_response_from_records(
                    self.resource.schema(exclude=tuple(self.resource.obj_exclude), only=tuple(self.resource.obj_only))
                        .dump(objects.items, many=True).data, 'csv', 200,  self.resource.model.__name__)

            resources = objects.paginate(page=self.resource.page, per_page=self.resource.limit)
            if resources.items:
                return make_response(jsonify({'success': True,
                                              'data': self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                           only=tuple(self.resource.obj_only))
                                             .dump(resources.items, many=True).data, 'total': resources.total}), 200)
            return make_response(jsonify({'error': True, 'message': 'No Resource Found'}), 404)
api.py 文件源码 项目:SuperFlaskSeed 作者: saurabh1e 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def patch(self, slug):
        obj = self.resource.model.query.get(slug)
        if not obj:
            return make_response(jsonify({'error': True, 'message': 'Resource not found'}), 404)
        try:
            data, status = self.resource.patch_resource(obj)
        except (SQLIntegrityError, SQlOperationalError) as e:
            db.session.rollback()
            e.message['error'] = True
            return make_response(jsonify(e.message), e.status)
        return make_response(jsonify(data), status)
api.py 文件源码 项目:SuperFlaskSeed 作者: saurabh1e 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def delete(self, slug):

        obj = self.resource.model.query.get(slug)
        if obj:
            if self.resource.has_delete_permission(obj):
                db.session.delete(obj)
                db.session.commit()
                return make_response(jsonify({}), 204)
            else:
                return make_response(
                    jsonify({'error': True, 'message': 'Forbidden Permission Denied To Delete Resource'}), 403)
        return make_response(jsonify({'error': True, 'message': 'Resource not found'}), 404)
resources.py 文件源码 项目:zou 作者: cgwire 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def get(self, shot_id):
        """
        Resource to retrieve the casting of a given shot.
        """
        shot = shots_service.get_shot(shot_id)
        if not permissions.has_manager_permissions():
            user_service.check_has_task_related(shot["project_id"])
        return breakdown_service.get_casting(shot_id)
resources.py 文件源码 项目:zou 作者: cgwire 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def put(self, shot_id):
        """
        Resource to allow the modification of assets linked to a shot.
        """
        casting = request.json
        permissions.check_manager_permissions()
        return breakdown_service.update_casting(shot_id, casting)
__init__.py 文件源码 项目:Canella-CMS 作者: mush42 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def delete(self, pk):
        obj = self.query_one(pk)
        self.abort(obj)
        db.session.delete(obj)
        db.session.commit()
        return {'message': 'Resource was deleted successfully'}
__init__.py 文件源码 项目:Canella-CMS 作者: mush42 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def abort(self, obj, status_code=404, message='Resource was not found'):
        if obj is None:
            flask_abort(status_code, message)
resource.py 文件源码 项目:peach 作者: sebastiandev 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def __init__(self, *args, **kwargs):
        BaseResource.__init__(self, FlaskRequestHelper(), *args, **kwargs)
        flask_restful.Resource.__init__(self)
        self._request_helper.parse(flask_restful.request, self.REQUEST_ARGS)

    # Need to specify these again because for some reason Flask MethodView doesn't pick up
    # the class methods inherited from BaseResource
api.py 文件源码 项目:open-pos-api 作者: saurabh1e 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def get(self, slug=None):
        if slug:
            obj = self.resource.model.query.filter(self.resource.model.id == slug)
            obj = self.resource.has_read_permission(obj).first()
            if obj:
                return make_response(jsonify(self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                  only=tuple(self.resource.obj_only)).dump(
                    obj, many=False).data), 200)

            return make_response(jsonify({'error': True, 'message': 'Resource not found'}), 404)

        else:
            objects = self.resource.apply_filters(queryset=self.resource.model.query, **request.args)
            objects = self.resource.has_read_permission(objects)

            if '__order_by' in request.args:
                objects = self.resource.apply_ordering(objects, request.args['__order_by'])

            if '__export__' in request.args and self.resource.export is True:
                objects = objects.paginate(page=self.resource.page, per_page=self.resource.max_export_limit)
                return make_response_from_records(
                    self.resource.schema(exclude=tuple(self.resource.obj_exclude), only=tuple(self.resource.obj_only))
                        .dump(objects.items, many=True).data, 'csv', 200,  self.resource.model.__name__)

            resources = objects.paginate(page=self.resource.page, per_page=self.resource.limit)
            if resources.items:
                return make_response(jsonify({'success': True,
                                              'data': self.resource.schema(exclude=tuple(self.resource.obj_exclude),
                                                                           only=tuple(self.resource.obj_only))
                                             .dump(resources.items, many=True).data, 'total': resources.total}), 200)
            return make_response(jsonify({'error': True, 'message': 'No Resource Found'}), 404)


问题


面经


文章

微信
公众号

扫码关注公众号