PK{I#tests/test_sqlalchemy_data_layer.py# -*- coding: utf-8 -*- from six.moves.urllib.parse import urlencode import pytest import json import datetime from sqlalchemy import create_engine, Column, Integer, DateTime, String from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from flask import Blueprint from marshmallow_jsonapi.flask import Schema from marshmallow_jsonapi import fields from flask_rest_jsonapi import ResourceList, ResourceDetail, SqlalchemyDataLayer @pytest.fixture(scope="session") def base(): yield declarative_base() @pytest.fixture(scope="session") def item_cls(base): class Item(base): __tablename__ = 'item' id = Column(Integer, primary_key=True) title = Column(String) content = Column(String) created = Column(DateTime, default=datetime.datetime.utcnow) yield Item @pytest.fixture(scope="session") def engine(item_cls): engine = create_engine("sqlite:///:memory:") item_cls.metadata.create_all(engine) return engine @pytest.fixture(scope="session") def session(engine): Session = sessionmaker(bind=engine) return Session() @pytest.fixture(scope="session") def base_query(item_cls): def get_base_query(self, **view_kwargs): return self.session.query(item_cls) yield get_base_query @pytest.fixture(scope="session") def dummy_decorator(): def deco(f): def wrapper_f(*args, **kwargs): return f(*args, **kwargs) return wrapper_f yield deco @pytest.fixture(scope="session") def item_schema(): class ItemSchema(Schema): class Meta: type_ = 'item' self_view = 'rest_api.item_detail' self_view_kwargs = {'item_id': ''} self_view_many = 'rest_api.item_list' id = fields.Str(dump_only=True) title = fields.Str() content = fields.Str() created = fields.DateTime() yield ItemSchema @pytest.fixture(scope="session") def item_list_resource(session, item_cls, base_query, dummy_decorator, item_schema): class ItemList(ResourceList): class Meta: data_layer = {'cls': SqlalchemyDataLayer, 'kwargs': {'model': item_cls, 'session': session}, 'get_base_query': base_query} get_decorators = [dummy_decorator] post_decorators = [dummy_decorator] resource_type = 'item' schema = {'cls': item_schema} endpoint = {'alias': 'rest_api.item_list'} yield ItemList @pytest.fixture(scope="session") def item_detail_resource(session, item_cls, base_query, dummy_decorator, item_schema): class ItemDetail(ResourceDetail): class Meta: data_layer = {'cls': SqlalchemyDataLayer, 'kwargs': {'model': item_cls, 'session': session, 'id_field': 'id', 'url_param_name': 'item_id'}, 'get_base_query': base_query} get_decorators = [dummy_decorator] patch_decorators = [dummy_decorator] delete_decorators = [dummy_decorator] resource_type = 'item' schema = {'cls': item_schema} yield ItemDetail @pytest.fixture(scope="session") def item_list_resource_not_allowed(session, item_cls, base_query, dummy_decorator, item_schema): class ItemList(ResourceList): class Meta: data_layer = {'cls': SqlalchemyDataLayer, 'kwargs': {'model': item_cls, 'session': session}, 'get_base_query': base_query} get_decorators = [dummy_decorator] not_allowed_methods = ['POST'] resource_type = 'item' schema_cls = item_schema collection_endpoint = 'rest_api.item_list' yield ItemList @pytest.fixture(scope="session") def rest_api_blueprint(client): bp = Blueprint('rest_api', __name__) yield bp @pytest.fixture(scope="session") def register_routes(client, rest_api_blueprint, item_list_resource, item_detail_resource, item_list_resource_not_allowed): rest_api_blueprint.add_url_rule('/items', view_func=item_list_resource.as_view('item_list')) rest_api_blueprint.add_url_rule('/items/', view_func=item_detail_resource.as_view('item_detail')) rest_api_blueprint.add_url_rule('/items_not_allowed', view_func=item_list_resource_not_allowed.as_view('item_list_not_allowed')) client.application.register_blueprint(rest_api_blueprint) def test_get_list_resource(client, register_routes): querystring = urlencode({'page[number]': 3, 'page[size]': 1, 'fields[item]': 'title,content', 'sort': '-created,title', 'filter[item]': json.dumps([{'field': 'created', 'op': 'gt', 'value': '2016-11-10'}])}) response = client.get('/items' + '?' + querystring, content_type='application/vnd.api+json') assert response.status_code == 200 def test_post_list_resource(client, register_routes): response = client.post('/items', data=json.dumps({"data": {"type": "item", "attributes": {"title": "test"}}}), content_type='application/vnd.api+json') assert response.status_code == 201 def test_get_detail_resource(client, register_routes): response = client.get('/items/1', content_type='application/vnd.api+json') assert response.status_code == 200 def test_patch_patch_resource(client, register_routes): response = client.patch('/items/1', data=json.dumps({"data": {"type": "item", "id": 1, "attributes": {"title": "test2"}}}), content_type='application/vnd.api+json') assert response.status_code == 200 def test_delete_detail_resource(client, register_routes): response = client.delete('/items/1', content_type='application/vnd.api+json') assert response.status_code == 204 def test_post_list_resource_not_allowed(client, register_routes): response = client.post('/items_not_allowed', data=json.dumps({"data": {"type": "item", "attributes": {"title": "test"}}}), content_type='application/vnd.api+json') assert response.status_code == 405 def test_get_detail_resource_not_found(client, register_routes): response = client.get('/items/2', content_type='application/vnd.api+json') assert response.status_code == 404 def test_patch_patch_resource_error(client, register_routes): response = client.patch('/items/1', data=json.dumps({"data": {"type": "item", "attributes": {"title": "test2"}}}), content_type='application/vnd.api+json') assert response.status_code == 422 def test_wrong_content_type(client, register_routes): response = client.delete('/items/1') assert response.status_code == 415 def test_response_content_type(client, register_routes): response = client.delete('/items/1', content_type='application/vnd.api+json') assert response.headers['Content-Type'] == 'application/vnd.api+json' PKhrIQ  tests/conftest.py# -*- coding: utf-8 -*- import pytest from flask import Flask @pytest.fixture(scope="session") def app(): app = Flask(__name__) return app @pytest.yield_fixture(scope="session") def client(app): with app.test_client() as client: yield client PK-rItests/__init__.pyPKhJ1!flask_rest_jsonapi/querystring.py# -*- coding: utf-8 -*- import json from flask_rest_jsonapi.exceptions import BadRequest, InvalidFilters, InvalidSort class QueryStringManager(object): """Querystring parser according to jsonapi reference """ MANAGED_KEYS = ( 'filter', 'page', 'fields', 'sort', 'include' ) def __init__(self, querystring, schema): """Initialization instance :param dict querystring: query string dict from request.args """ if not isinstance(querystring, dict): raise ValueError('QueryStringManager require a dict-like object query_string parameter') self.qs = querystring self.schema = schema def _get_key_values(self, name): """Return a dict containing key / values items for a given key, used for items like filters, page, etc. :param str name: name of the querystring parameter :return dict: a dict of key / values items """ results = {} for key, value in self.qs.items(): try: if not key.startswith(name): continue key_start = key.index('[') + 1 key_end = key.index(']') item_key = key[key_start:key_end] if ',' in value: item_value = value.split(',') else: item_value = value results.update({item_key: item_value}) except Exception: raise BadRequest({'parameter': key}, "Parse error") return results @property def querystring(self): """Return original querystring but containing only managed keys :return dict: dict of managed querystring parameter """ return {key: value for (key, value) in self.qs.items() if key.startswith(self.MANAGED_KEYS)} @property def filters(self): """Return filters from query string. :return list: filter information """ filters = self.qs.get('filter') if filters is not None: try: filters = json.loads(filters) except (ValueError, TypeError): raise InvalidFilters("Parse error") return filters @property def pagination(self): """Return all page parameters as a dict. :return dict: a dict of pagination information To allow multiples strategies, all parameters starting with `page` will be included. e.g:: { "number": '25', "size": '150', } Example with number strategy:: >>> query_string = {'page[number]': '25', 'page[size]': '10'} >>> parsed_query.pagination {'number': '25', 'size': '10'} """ # check values type result = self._get_key_values('page') for key, value in result.items(): if key not in ('number', 'size'): raise BadRequest({'parameter': 'page'}, "{} is not a valid parameter of pagination".format(key)) try: int(value) except ValueError: raise BadRequest({'parameter': 'page[{}]'.format(key)}, "Parse error") return result @property def fields(self): """Return fields wanted by client. :return dict: a dict of sparse fieldsets information Return value will be a dict containing all fields by resource, for example:: { "user": ['name', 'email'], } """ result = self._get_key_values('fields') for key, value in result.items(): if not isinstance(value, list): result[key] = [value] return result @property def sorting(self): """Return fields to sort by including sort name for SQLAlchemy and row sort parameter for other ORMs :return list: a list of sorting information Example of return value:: [ {'field': 'created_at', 'order': 'desc'}, ] """ if self.qs.get('sort'): sorting_results = [] for sort_field in self.qs['sort'].split(','): field = sort_field.replace('-', '') if field not in self.schema._declared_fields: raise InvalidSort("{} has no attribut {}".format(self.schema.__name__, field)) if self.schema._declared_fields[field].attribute is not None: field = self.schema._declared_fields[field].attribute order = 'desc' if sort_field.startswith('-') else 'asc' sorting_results.append({'field': field, 'order': order}) return sorting_results return [] @property def include(self): """Return fields to include :return list: a list of include information """ include_param = self.qs.get('include') return include_param.split(',') if include_param else [] PKJmJ:nnflask_rest_jsonapi/errors.py# -*- coding: utf-8 -*- def jsonapi_errors(jsonapi_errors): """Construct api error according to jsonapi 1.0 :param iterable jsonapi_errors: an iterable of jsonapi error :return dict: a dict of errors according to jsonapi 1.0 """ return {'errors': [jsonapi_error for jsonapi_error in jsonapi_errors], 'jsonapi': {'version': '1.0'}} PKmWJRƐ__ flask_rest_jsonapi/exceptions.py# -*- coding: utf-8 -*- class JsonApiException(Exception): title = 'Unknow error' status = 500 def __init__(self, source, detail, title=None, status=None): """Initialize a jsonapi exception :param dict source: the source of the error :param str detail: the detail of the error """ self.source = source self.detail = detail if title is not None: self.title = title if status is not None: self.status = status def to_dict(self): return {'status': self.status, 'source': self.source, 'title': self.title, 'detail': self.detail} class BadRequest(JsonApiException): title = "Bad request" status = 400 class InvalidField(BadRequest): title = "Invalid fields querystring parameter." def __init__(self, detail): self.source = {'parameter': 'fields'} self.detail = detail class InvalidInclude(BadRequest): title = "Invalid include querystring parameter." def __init__(self, detail): self.source = {'parameter': 'include'} self.detail = detail class InvalidFilters(BadRequest): title = "Invalid filters querystring parameter." def __init__(self, detail): self.source = {'parameter': 'filters'} self.detail = detail class InvalidSort(BadRequest): title = "Invalid sort querystring parameter." def __init__(self, detail): self.source = {'parameter': 'sort'} self.detail = detail class ObjectNotFound(JsonApiException): title = "Object not found" status = 404 class RelatedObjectNotFound(ObjectNotFound): title = "Related object not found" class RelationNotFound(JsonApiException): title = "Relation not found" class InvalidType(JsonApiException): title = "Invalid type" status = 409 PKjJIEEEflask_rest_jsonapi/api.py# -*- coding: utf-8 -*- from flask import Blueprint class Api(object): def __init__(self, app=None): self.app = None self.blueprint = None if app is not None: if isinstance(app, Blueprint): self.blueprint = app else: self.app = app self.resources = [] def init_app(self, app=None): """Update flask application with our api :param Application app: a flask application """ if self.app is None: self.app = app if self.blueprint is not None: self.app.register_blueprint(self.blueprint) else: for resource in self.resources: self.route(**resource) def route(self, resource, view, *urls, **kwargs): """Create an api view. :param Resource resource: a resource class inherited from flask_rest_jsonapi.resource.Resource :param str view: the view name :param list urls: the urls of the view :param dict kwargs: additional options of the route """ resource.view = view view_func = resource.as_view(view) url_rule_options = kwargs.get('url_rule_options') or dict() if self.app is not None: for url in urls: self.app.add_url_rule(url, view_func=view_func, **url_rule_options) elif self.blueprint is not None: resource.view = '.'.join([self.blueprint.name, resource.view]) for url in urls: self.blueprint.add_url_rule(url, view_func=view_func, **url_rule_options) else: self.resources.append({'resource': resource, 'view': view, 'urls': urls, 'url_rule_options': url_rule_options}) PKhJ0 flask_rest_jsonapi/schema.py# -*- coding: utf-8 -*- from marshmallow import class_registry from marshmallow.base import SchemaABC from marshmallow_jsonapi.fields import Relationship from flask_rest_jsonapi.exceptions import InvalidField, InvalidInclude def compute_schema(schema_cls, default_kwargs, qs, include): """Compute a schema around compound documents and sparse fieldsets :param Schema schema_cls: the schema class :param dict default_kwargs: the schema default kwargs :param QueryStringManager qs: qs :param list include: the relation field to include data from :return Schema schema: the schema computed """ # manage include_data parameter of the schema schema_kwargs = default_kwargs schema_kwargs['include_data'] = tuple() if include: for include_path in include: field = include_path.split('.')[0] if field not in schema_cls._declared_fields: raise InvalidInclude("{} has no attribut {}".format(schema_cls.__name__, field)) elif not isinstance(schema_cls._declared_fields[field], Relationship): raise InvalidInclude("{} is not a relationship attribut of {}".format(field, schema_cls.__name__)) schema_kwargs['include_data'] += (field, ) # make sure id field is in only parameter unless marshamllow will raise an Exception if schema_kwargs.get('only') is not None and 'id' not in schema_kwargs['only']: schema_kwargs['only'] += ('id',) # create base schema instance schema = schema_cls(**schema_kwargs) # manage sparse fieldsets if schema.opts.type_ in qs.fields: # check that sparse fieldsets exists in the schema for field in qs.fields[schema.opts.type_]: if field not in schema.declared_fields: raise InvalidField("{} has no attribut {}".format(schema.__class__.__name__, field)) tmp_only = set(schema.declared_fields.keys()) & set(qs.fields[schema.opts.type_]) if schema.only: tmp_only &= set(schema.only) schema.only = tuple(tmp_only) # make sure again that id field is in only parameter unless marshamllow will raise an Exception if schema.only is not None and 'id' not in schema.only: schema.only += ('id',) # manage compound documents if include: for include_path in include: field = include_path.split('.')[0] relation_field = schema.declared_fields[field] related_schema_cls = schema.declared_fields[field].__dict__['_Relationship__schema'] related_schema_kwargs = {} if isinstance(related_schema_cls, SchemaABC): related_schema_kwargs['many'] = related_schema_cls.many related_schema_cls = related_schema_cls.__class__ if isinstance(related_schema_cls, str): related_schema_cls = class_registry.get_class(related_schema_cls) if '.' in include_path: related_include = ['.'.join(include_path.split('.')[1:])] else: related_include = None related_schema = compute_schema(related_schema_cls, related_schema_kwargs, qs, related_include) relation_field.__dict__['_Relationship__schema'] = related_schema return schema def get_relationships(schema): """Return relationship fields of a schema :param Schema schema: a marshmallow schema """ return [key for (key, value) in schema._declared_fields.items() if isinstance(value, Relationship)] PKJmJw_ flask_rest_jsonapi/pagination.py# -*- coding: utf-8 -*- from six.moves.urllib.parse import urlencode from math import ceil from copy import copy from flask_rest_jsonapi.constants import DEFAULT_PAGE_SIZE def add_pagination_links(data, object_count, querystring, base_url): """Add pagination links to result :param dict data: the result of the view :param int object_count: number of objects in result :param QueryStringManager querystring: the managed querystring fields and values :param str base_url: the base url for pagination """ links = {} all_qs_args = copy(querystring.querystring) links['self'] = base_url # compute self link if all_qs_args: links['self'] += '?' + urlencode(all_qs_args) if querystring.pagination.get('size') != '0' and object_count > 1: # compute last link page_size = int(querystring.pagination.get('size', 0)) or DEFAULT_PAGE_SIZE last_page = int(ceil(object_count / page_size)) if last_page > 1: links['first'] = links['last'] = base_url try: del all_qs_args['page[number]'] except KeyError: pass # compute first link if all_qs_args: links['first'] += '?' + urlencode(all_qs_args) all_qs_args.update({'page[number]': last_page}) links['last'] += '?' + urlencode(all_qs_args) # compute previous and next link current_page = int(querystring.pagination.get('number', 0)) or 1 if current_page > 1: all_qs_args.update({'page[number]': current_page - 1}) links['prev'] = '?'.join((base_url, urlencode(all_qs_args))) if current_page < last_page: all_qs_args.update({'page[number]': current_page + 1}) links['next'] = '?'.join((base_url, urlencode(all_qs_args))) data['links'] = links PKhrIcYYflask_rest_jsonapi/constants.py# -*- coding: utf-8 -*- # default number of items for pagination DEFAULT_PAGE_SIZE = 20 PKJmJ>y. . flask_rest_jsonapi/decorators.py# -*- coding: utf-8 -*- import json from flask import request, make_response from flask_rest_jsonapi.errors import jsonapi_errors def check_headers(f): """Check headers according to jsonapi reference :param callable f: the function to decorate :return callable: the wrapped function """ def wrapped_f(*args, **kwargs): if request.method in ('POST', 'PATCH'): if request.headers['Content-Type'] != 'application/vnd.api+json': error = json.dumps(jsonapi_errors([{'source': '', 'detail': "Content-Type header must be application/vnd.api+json", 'title': 'InvalidRequestHeader', 'status': 415}])) return make_response(error, 415, {'Content-Type': 'application/vnd.api+json'}) if request.headers.get('Accept') and request.headers['Accept'] != 'application/vnd.api+json': error = json.dumps(jsonapi_errors([{'source': '', 'detail': "Accept header must be application/vnd.api+json", 'title': 'InvalidRequestHeader', 'status': 406}])) return make_response(error, 406, {'Content-Type': 'application/vnd.api+json'}) return f(*args, **kwargs) return wrapped_f def check_method_requirements(f): """Check methods requirements :param callable f: the function to decorate :return callable: the wrapped function """ def wrapped_f(self, *args, **kwargs): cls = type(self) error_message = "You must provide {error_field} in {cls} to get access to the default {method} method" error_data = {'cls': cls.__name__, 'method': request.method} if not hasattr(self, '_data_layer'): error_data.update({'error_field': 'a data layer class'}) raise Exception(error_message.format(**error_data)) if request.method != 'DELETE': if not hasattr(self, 'schema'): error_data.update({'error_field': 'a schema class'}) raise Exception(error_message.format(**error_data)) return f(self, *args, **kwargs) return wrapped_f PKVcJO\MMflask_rest_jsonapi/__init__.py# -*- coding: utf-8 -*- from flask_rest_jsonapi.api import Api from flask_rest_jsonapi.resource import ResourceList, ResourceDetail, ResourceRelationship from flask_rest_jsonapi.exceptions import JsonApiException __all__ = [ 'Api', 'ResourceList', 'ResourceDetail', 'ResourceRelationship', 'JsonApiException' ] PKJmJJJflask_rest_jsonapi/resource.py# -*- coding: utf-8 -*- import inspect import json from copy import copy from werkzeug.wrappers import Response from flask import request, url_for, make_response, current_app from flask.views import MethodView from marshmallow_jsonapi.exceptions import IncorrectTypeError from marshmallow import ValidationError from flask_rest_jsonapi.errors import jsonapi_errors from flask_rest_jsonapi.querystring import QueryStringManager as QSManager from flask_rest_jsonapi.pagination import add_pagination_links from flask_rest_jsonapi.exceptions import InvalidType, BadRequest, JsonApiException, RelationNotFound from flask_rest_jsonapi.decorators import check_headers, check_method_requirements from flask_rest_jsonapi.schema import compute_schema, get_relationships from flask_rest_jsonapi.data_layers.base import BaseDataLayer from flask_rest_jsonapi.data_layers.alchemy import SqlalchemyDataLayer class Resource(MethodView): def __new__(cls): if hasattr(cls, 'data_layer'): if not isinstance(cls.data_layer, dict): raise Exception("You must provide a data layer information as dict in {}".format(cls.__name__)) if cls.data_layer.get('class') is not None and BaseDataLayer not in inspect.getmro(cls.data_layer['class']): raise Exception("You must provide a data layer class inherited from BaseDataLayer in {}" .format(cls.__name__)) data_layer_cls = cls.data_layer.get('class', SqlalchemyDataLayer) cls._data_layer = data_layer_cls(**cls.data_layer) cls._data_layer.resource = cls for method in getattr(cls, 'methods', ('GET', 'POST', 'PATCH', 'DELETE')): if hasattr(cls, method.lower()): setattr(cls, method.lower(), check_headers(getattr(cls, method.lower()))) for method in ('get', 'post', 'patch', 'delete'): if hasattr(cls, '{}_decorators'.format(method)) and hasattr(cls, method): for decorator in getattr(cls, '{}_decorators'.format(method)): setattr(cls, method, decorator(getattr(cls, method))) return super(Resource, cls).__new__(cls) def dispatch_request(self, *args, **kwargs): method = getattr(self, request.method.lower(), None) if method is None and request.method == 'HEAD': method = getattr(self, 'get', None) assert method is not None, 'Unimplemented method {}'.format(request.method) headers = {'Content-Type': 'application/vnd.api+json'} try: response = method(*args, **kwargs) except JsonApiException as e: return make_response(json.dumps(jsonapi_errors([e.to_dict()])), e.status, headers) except Exception as e: if current_app.config['DEBUG'] is True: raise e exc = JsonApiException('', str(e)) return make_response(json.dumps(jsonapi_errors([exc.to_dict()])), exc.status, headers) if isinstance(response, Response): response.headers.add('Content-Type', 'application/vnd.api+json') return response if not isinstance(response, tuple): if isinstance(response, dict): response.update({'jsonapi': {'version': '1.0'}}) return make_response(json.dumps(response), 200, headers) try: data, status_code, headers = response headers.update({'Content-Type': 'application/vnd.api+json'}) except ValueError: pass try: data, status_code = response except ValueError: pass if isinstance(data, dict): data.update({'jsonapi': {'version': '1.0'}}) return make_response(json.dumps(data), status_code, headers) class ResourceList(Resource): @check_method_requirements def get(self, *args, **kwargs): """Retrieve a collection of objects """ self.before_get(*args, **kwargs) qs = QSManager(request.args, self.schema) object_count, objects = self._data_layer.get_collection(qs, **kwargs) schema_kwargs = getattr(self, 'get_schema_kwargs', dict()) schema_kwargs.update({'many': True}) schema = compute_schema(self.schema, schema_kwargs, qs, qs.include) result = schema.dump(objects).data view_kwargs = request.view_args if getattr(self, 'view_kwargs', None) is True else dict() add_pagination_links(result, object_count, qs, url_for(self.view, **view_kwargs)) self.after_get(result) return result @check_method_requirements def post(self, *args, **kwargs): """Create an object """ self.before_post(*args, **kwargs) json_data = request.get_json() qs = QSManager(request.args, self.schema) schema = compute_schema(self.schema, getattr(self, 'post_schema_kwargs', dict()), qs, qs.include) try: data, errors = schema.load(json_data) except IncorrectTypeError as e: errors = e.messages for error in errors['errors']: error['status'] = '409' error['title'] = "Incorrect type" return errors, 409 except ValidationError as e: errors = e.messages for message in errors['errors']: message['status'] = '422' message['title'] = "Validation error" return errors, 422 if errors: for error in errors['errors']: error['status'] = "422" error['title'] = "Validation error" return errors, 422 obj = self._data_layer.create_object(data, **kwargs) result = schema.dump(obj).data self.after_post(result) return result, 201, {'Location': result['data']['links']['self']} def before_get(self, *args, **kwargs): pass def after_get(self, result): pass def before_post(self, *args, **kwargs): pass def after_post(self, result): pass class ResourceDetail(Resource): @check_method_requirements def get(self, *args, **kwargs): """Get object details """ self.before_get(*args, **kwargs) obj = self._data_layer.get_object(**kwargs) qs = QSManager(request.args, self.schema) schema = compute_schema(self.schema, getattr(self, 'get_schema_kwargs', dict()), qs, qs.include) result = schema.dump(obj).data self.after_get(result) return result @check_method_requirements def patch(self, *args, **kwargs): """Update an object """ self.before_patch(*args, **kwargs) json_data = request.get_json() qs = QSManager(request.args, self.schema) schema_kwargs = getattr(self, 'patch_schema_kwargs', dict()) schema_kwargs.update({'partial': True}) schema = compute_schema(self.schema, schema_kwargs, qs, qs.include) try: data, errors = schema.load(json_data) except IncorrectTypeError as e: errors = e.messages for error in errors['errors']: error['status'] = '409' error['title'] = "Incorrect type" return errors, 409 except ValidationError as e: errors = e.messages for message in errors['errors']: message['status'] = '422' message['title'] = "Validation error" return errors, 422 if errors: for error in errors['errors']: error['status'] = "422" error['title'] = "Validation error" return errors, 422 if 'id' not in json_data['data']: raise BadRequest('/data/id', 'Missing id in "data" node') if json_data['data']['id'] != str(kwargs[self.data_layer.get('url_field', 'id')]): raise BadRequest('/data/id', 'Value of id does not match the resource identifier in url') obj = self._data_layer.get_object(**kwargs) self._data_layer.update_object(obj, data, **kwargs) result = schema.dump(obj).data self.after_patch(result) return result @check_method_requirements def delete(self, *args, **kwargs): """Delete an object """ self.before_delete(*args, **kwargs) obj = self._data_layer.get_object(**kwargs) self._data_layer.delete_object(obj, **kwargs) result = {'meta': 'Object successful deleted'} self.after_delete(result) return result def before_get(self, *args, **kwargs): pass def after_get(self, result): pass def before_patch(self, *args, **kwargs): pass def after_patch(self, result): pass def before_delete(self, *args, **kwargs): pass def after_delete(self, result): pass class ResourceRelationship(Resource): @check_method_requirements def get(self, *args, **kwargs): """Get a relationship details """ self.before_get(*args, **kwargs) relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() related_view = self.schema._declared_fields[relationship_field].related_view related_view_kwargs = self.schema._declared_fields[relationship_field].related_view_kwargs obj, data = self._data_layer.get_relationship(model_relationship_field, related_type_, related_id_field, **kwargs) for key, value in copy(related_view_kwargs).items(): if isinstance(value, str) and value.startswith('<') and value.endswith('>'): tmp_obj = obj for field in value[1:-1].split('.'): tmp_obj = getattr(tmp_obj, field) related_view_kwargs[key] = tmp_obj result = {'links': {'self': request.base_url, 'related': url_for(related_view, **related_view_kwargs)}, 'data': data} qs = QSManager(request.args, self.schema) if qs.include: schema = compute_schema(self.schema, dict(), qs, qs.include) serialized_obj = schema.dump(obj) result['included'] = serialized_obj.data.get('included', dict()) self.after_get(result) return result @check_method_requirements def post(self, *args, **kwargs): """Add / create relationship(s) """ self.before_post(*args, **kwargs) json_data = request.get_json() relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() if 'data' not in json_data: raise BadRequest('/data', 'You must provide data with a "data" route node') if isinstance(json_data['data'], dict): if 'type' not in json_data['data']: raise BadRequest('/data/type', 'Missing type in "data" node') if 'id' not in json_data['data']: raise BadRequest('/data/id', 'Missing id in "data" node') if json_data['data']['type'] != related_type_: raise InvalidType('/data/type', 'The type field does not match the resource type') if isinstance(json_data['data'], list): for obj in json_data['data']: if 'type' not in obj: raise BadRequest('/data/type', 'Missing type in "data" node') if 'id' not in obj: raise BadRequest('/data/id', 'Missing id in "data" node') if obj['type'] != related_type_: raise InvalidType('/data/type', 'The type provided does not match the resource type') obj_, updated = self._data_layer.create_relationship(json_data, model_relationship_field, related_id_field, **kwargs) qs = QSManager(request.args, self.schema) includes = qs.include if relationship_field not in qs.include: includes.append(relationship_field) schema = compute_schema(self.schema, dict(), qs, includes) status_code = 200 if updated is True else 204 result = schema.dump(obj_).data if result.get('links', {}).get('self') is not None: result['links']['self'] = request.base_url self.after_post(result) return result, status_code @check_method_requirements def patch(self, *args, **kwargs): """Update a relationship """ self.before_patch(*args, **kwargs) json_data = request.get_json() relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() if 'data' not in json_data: raise BadRequest('/data', 'You must provide data with a "data" route node') if isinstance(json_data['data'], dict): if 'type' not in json_data['data']: raise BadRequest('/data/type', 'Missing type in "data" node') if 'id' not in json_data['data']: raise BadRequest('/data/id', 'Missing id in "data" node') if json_data['data']['type'] != related_type_: raise InvalidType('/data/type', 'The type field does not match the resource type') if isinstance(json_data['data'], list): for obj in json_data['data']: if 'type' not in obj: raise BadRequest('/data/type', 'Missing type in "data" node') if 'id' not in obj: raise BadRequest('/data/id', 'Missing id in "data" node') if obj['type'] != related_type_: raise InvalidType('/data/type', 'The type provided does not match the resource type') obj_, updated = self._data_layer.update_relationship(json_data, model_relationship_field, related_id_field, **kwargs) qs = QSManager(request.args, self.schema) includes = qs.include if relationship_field not in qs.include: includes.append(relationship_field) schema = compute_schema(self.schema, dict(), qs, includes) status_code = 200 if updated is True else 204 result = schema.dump(obj_).data if result.get('links', {}).get('self') is not None: result['links']['self'] = request.base_url self.after_patch(result) return result, status_code @check_method_requirements def delete(self, *args, **kwargs): """Delete relationship(s) """ self.before_delete(*args, **kwargs) json_data = request.get_json() relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() if 'data' not in json_data: raise BadRequest('/data', 'You must provide data with a "data" route node') if isinstance(json_data['data'], dict): if 'type' not in json_data['data']: raise BadRequest('/data/type', 'Missing type in "data" node') if 'id' not in json_data['data']: raise BadRequest('/data/id', 'Missing id in "data" node') if json_data['data']['type'] != related_type_: raise InvalidType('/data/type', 'The type field does not match the resource type') if isinstance(json_data['data'], list): for obj in json_data['data']: if 'type' not in obj: raise BadRequest('/data/type', 'Missing type in "data" node') if 'id' not in obj: raise BadRequest('/data/id', 'Missing id in "data" node') if obj['type'] != related_type_: raise InvalidType('/data/type', 'The type provided does not match the resource type') obj_, updated = self._data_layer.delete_relationship(json_data, model_relationship_field, related_id_field, **kwargs) qs = QSManager(request.args, self.schema) includes = qs.include if relationship_field not in qs.include: includes.append(relationship_field) schema = compute_schema(self.schema, dict(), qs, includes) status_code = 200 if updated is True else 204 result = schema.dump(obj_).data if result.get('links', {}).get('self') is not None: result['links']['self'] = request.base_url self.after_delete(result) return result, status_code def _get_relationship_data(self): """Get useful data for relationship management """ relationship_field = request.base_url.split('/')[-1] if relationship_field not in get_relationships(self.schema): raise RelationNotFound('', "{} has no attribut {}".format(self.schema.__name__, relationship_field)) related_type_ = self.schema._declared_fields[relationship_field].type_ related_id_field = self.schema._declared_fields[relationship_field].id_field if hasattr(self, 'schema_to_model') and self.schema_to_model.get(relationship_field) is not None: model_relationship_field = self.schema_to_model[relationship_field] else: model_relationship_field = relationship_field return relationship_field, model_relationship_field, related_type_, related_id_field def before_get(self, *args, **kwargs): pass def after_get(self, result): pass def before_post(self, *args, **kwargs): pass def after_post(self, result): pass def before_patch(self, *args, **kwargs): pass def after_patch(self, result): pass def before_delete(self, *args, **kwargs): pass def after_delete(self, result): pass PKJmJ@y. . Yflask_rest_jsonapi/decorators.pyPKVcJO\MMcflask_rest_jsonapi/__init__.pyPKJmJJJdflask_rest_jsonapi/resource.pyPKJmJ@