diff --git a/.gitignore b/.gitignore index 72364f9..5bff718 100644 --- a/.gitignore +++ b/.gitignore @@ -87,3 +87,8 @@ ENV/ # Rope project settings .ropeproject + +# IDE +.vscode/ +app.py + diff --git a/examples/api.py b/examples/api.py index 781f470..a76917a 100644 --- a/examples/api.py +++ b/examples/api.py @@ -1,21 +1,22 @@ # -*- coding: utf-8 -*- from flask import Flask -from flask_rest_jsonapi import Api, ResourceDetail, ResourceList, ResourceRelationship -from flask_rest_jsonapi.exceptions import ObjectNotFound from flask_sqlalchemy import SQLAlchemy -from sqlalchemy.orm.exc import NoResultFound -from marshmallow_jsonapi.flask import Schema, Relationship from marshmallow_jsonapi import fields +from marshmallow_jsonapi.flask import Relationship, Schema +from sqlalchemy.orm.exc import NoResultFound + +from flask_rest_jsonapi import Api, ResourceDetail, ResourceList, ResourceRelationship +from flask_rest_jsonapi.exceptions import ObjectNotFound # Create the Flask application app = Flask(__name__) -app.config['DEBUG'] = True -app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False +app.config["DEBUG"] = True +app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False # Initialize SQLAlchemy -app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db' +app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:////tmp/test.db" db = SQLAlchemy(app) @@ -31,8 +32,9 @@ class Person(db.Model): class Computer(db.Model): id = db.Column(db.Integer, primary_key=True) serial = db.Column(db.String) - person_id = db.Column(db.Integer, db.ForeignKey('person.id')) - person = db.relationship('Person', backref=db.backref('computers')) + person_id = db.Column(db.Integer, db.ForeignKey("person.id")) + person = db.relationship("Person", backref=db.backref("computers")) + db.create_all() @@ -40,120 +42,144 @@ class Computer(db.Model): # Create logical data abstraction (same as data storage for this first example) class PersonSchema(Schema): class Meta: - type_ = 'person' - self_view = 'person_detail' - self_view_kwargs = {'id': ''} - self_view_many = 'person_list' + type_ = "person" + self_view = "person_detail" + self_view_kwargs = {"id": ""} + self_view_many = "person_list" id = fields.Integer(as_string=True, dump_only=True) name = fields.Str(required=True, load_only=True) email = fields.Email(load_only=True) birth_date = fields.Date() - display_name = fields.Function(lambda obj: "{} <{}>".format(obj.name.upper(), obj.email)) - computers = Relationship(self_view='person_computers', - self_view_kwargs={'id': ''}, - related_view='computer_list', - related_view_kwargs={'id': ''}, - many=True, - schema='ComputerSchema', - type_='computer') + display_name = fields.Function( + lambda obj: "{} <{}>".format(obj.name.upper(), obj.email) + ) + computers = Relationship( + self_view="person_computers", + self_view_kwargs={"id": ""}, + related_view="computer_list", + related_view_kwargs={"id": ""}, + many=True, + schema="ComputerSchema", + type_="computer", + ) + table = Person class ComputerSchema(Schema): class Meta: - type_ = 'computer' - self_view = 'computer_detail' - self_view_kwargs = {'id': ''} + type_ = "computer" + self_view = "computer_detail" + self_view_kwargs = {"id": ""} id = fields.Integer(as_string=True, dump_only=True) serial = fields.Str(required=True) - owner = Relationship(attribute='person', - self_view='computer_person', - self_view_kwargs={'id': ''}, - related_view='person_detail', - related_view_kwargs={'computer_id': ''}, - schema='PersonSchema', - type_='person') + owner = Relationship( + attribute="person", + self_view="computer_person", + self_view_kwargs={"id": ""}, + related_view="person_detail", + related_view_kwargs={"computer_id": ""}, + schema="PersonSchema", + type_="person", + ) # Create resource managers class PersonList(ResourceList): schema = PersonSchema - data_layer = {'session': db.session, - 'model': Person} + data_layer = {"session": db.session, "model": Person} class PersonDetail(ResourceDetail): def before_get_object(self, view_kwargs): - if view_kwargs.get('computer_id') is not None: + if view_kwargs.get("computer_id") is not None: try: - computer = self.session.query(Computer).filter_by(id=view_kwargs['computer_id']).one() + computer = ( + self.session.query(Computer) + .filter_by(id=view_kwargs["computer_id"]) + .one() + ) except NoResultFound: - raise ObjectNotFound({'parameter': 'computer_id'}, - "Computer: {} not found".format(view_kwargs['computer_id'])) + raise ObjectNotFound( + {"parameter": "computer_id"}, + "Computer: {} not found".format(view_kwargs["computer_id"]), + ) else: if computer.person is not None: - view_kwargs['id'] = computer.person.id + view_kwargs["id"] = computer.person.id else: - view_kwargs['id'] = None + view_kwargs["id"] = None schema = PersonSchema - data_layer = {'session': db.session, - 'model': Person, - 'methods': {'before_get_object': before_get_object}} + data_layer = { + "session": db.session, + "model": Person, + "methods": {"before_get_object": before_get_object}, + } class PersonRelationship(ResourceRelationship): schema = PersonSchema - data_layer = {'session': db.session, - 'model': Person} + data_layer = {"session": db.session, "model": Person} class ComputerList(ResourceList): def query(self, view_kwargs): query_ = self.session.query(Computer) - if view_kwargs.get('id') is not None: + if view_kwargs.get("id") is not None: try: - self.session.query(Person).filter_by(id=view_kwargs['id']).one() + self.session.query(Person).filter_by(id=view_kwargs["id"]).one() except NoResultFound: - raise ObjectNotFound({'parameter': 'id'}, "Person: {} not found".format(view_kwargs['id'])) + raise ObjectNotFound( + {"parameter": "id"}, + "Person: {} not found".format(view_kwargs["id"]), + ) else: - query_ = query_.join(Person).filter(Person.id == view_kwargs['id']) + query_ = query_.join(Person).filter(Person.id == view_kwargs["id"]) return query_ def before_create_object(self, data, view_kwargs): - if view_kwargs.get('id') is not None: - person = self.session.query(Person).filter_by(id=view_kwargs['id']).one() - data['person_id'] = person.id + if view_kwargs.get("id") is not None: + person = self.session.query(Person).filter_by(id=view_kwargs["id"]).one() + data["person_id"] = person.id schema = ComputerSchema - data_layer = {'session': db.session, - 'model': Computer, - 'methods': {'query': query, - 'before_create_object': before_create_object}} + data_layer = { + "session": db.session, + "model": Computer, + "methods": {"query": query, "before_create_object": before_create_object}, + } class ComputerDetail(ResourceDetail): schema = ComputerSchema - data_layer = {'session': db.session, - 'model': Computer} + data_layer = {"session": db.session, "model": Computer} class ComputerRelationship(ResourceRelationship): schema = ComputerSchema - data_layer = {'session': db.session, - 'model': Computer} + data_layer = {"session": db.session, "model": Computer} # Create endpoints api = Api(app) -api.route(PersonList, 'person_list', '/persons') -api.route(PersonDetail, 'person_detail', '/persons/', '/computers//owner') -api.route(PersonRelationship, 'person_computers', '/persons//relationships/computers') -api.route(ComputerList, 'computer_list', '/computers', '/persons//computers') -api.route(ComputerDetail, 'computer_detail', '/computers/') -api.route(ComputerRelationship, 'computer_person', '/computers//relationships/owner') - -if __name__ == '__main__': +api.route(PersonList, "person_list", "/persons") +api.route( + PersonDetail, + "person_detail", + "/persons/", + "/computers//owner", +) +api.route( + PersonRelationship, "person_computers", "/persons//relationships/computers" +) +api.route(ComputerList, "computer_list", "/computers", "/persons//computers") +api.route(ComputerDetail, "computer_detail", "/computers/") +api.route( + ComputerRelationship, "computer_person", "/computers//relationships/owner" +) + +if __name__ == "__main__": # Start application app.run(debug=True) diff --git a/examples/api_nested.py b/examples/api_nested.py index 2b3779a..96caf2d 100644 --- a/examples/api_nested.py +++ b/examples/api_nested.py @@ -1,22 +1,23 @@ # -*- coding: utf-8 -*- from flask import Flask -from flask_rest_jsonapi import Api, ResourceDetail, ResourceList, ResourceRelationship -from flask_rest_jsonapi.exceptions import ObjectNotFound from flask_sqlalchemy import SQLAlchemy -from sqlalchemy.orm.exc import NoResultFound -from marshmallow_jsonapi.flask import Schema, Relationship from marshmallow import Schema as MarshmallowSchema from marshmallow_jsonapi import fields +from marshmallow_jsonapi.flask import Relationship, Schema +from sqlalchemy.orm.exc import NoResultFound + +from flask_rest_jsonapi import Api, ResourceDetail, ResourceList, ResourceRelationship +from flask_rest_jsonapi.exceptions import ObjectNotFound # Create the Flask application app = Flask(__name__) -app.config['DEBUG'] = True -app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False +app.config["DEBUG"] = True +app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False # Initialize SQLAlchemy -app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db' +app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:////tmp/test.db" db = SQLAlchemy(app) @@ -27,35 +28,43 @@ class Person(db.Model): email = db.Column(db.String) birth_date = db.Column(db.Date) password = db.Column(db.String) - tags = db.relationship("Person_Tag", cascade="save-update, merge, delete, delete-orphan") - single_tag = db.relationship("Person_Single_Tag", uselist=False, cascade="save-update, merge, delete, delete-orphan") + tags = db.relationship( + "Person_Tag", cascade="save-update, merge, delete, delete-orphan" + ) + single_tag = db.relationship( + "Person_Single_Tag", + uselist=False, + cascade="save-update, merge, delete, delete-orphan", + ) json_tags = db.Column(db.JSON) class Computer(db.Model): id = db.Column(db.Integer, primary_key=True) serial = db.Column(db.String) - person_id = db.Column(db.Integer, db.ForeignKey('person.id')) - person = db.relationship('Person', backref=db.backref('computers')) + person_id = db.Column(db.Integer, db.ForeignKey("person.id")) + person = db.relationship("Person", backref=db.backref("computers")) class Person_Tag(db.Model): - id = db.Column(db.Integer, db.ForeignKey('person.id'), primary_key=True, index=True) + id = db.Column(db.Integer, db.ForeignKey("person.id"), primary_key=True, index=True) key = db.Column(db.String, primary_key=True) value = db.Column(db.String, primary_key=True) class Person_Single_Tag(db.Model): - id = db.Column(db.Integer, db.ForeignKey('person.id'), primary_key=True, index=True) + id = db.Column(db.Integer, db.ForeignKey("person.id"), primary_key=True, index=True) key = db.Column(db.String) value = db.Column(db.String) + db.create_all() + # Create schema class PersonTagSchema(MarshmallowSchema): class Meta: - type_ = 'person_tag' + type_ = "person_tag" id = fields.Str(dump_only=True, load_only=True) key = fields.Str() @@ -64,7 +73,7 @@ class Meta: class PersonSingleTagSchema(MarshmallowSchema): class Meta: - type_ = 'person_single_tag' + type_ = "person_single_tag" id = fields.Str(dump_only=True, load_only=True) key = fields.Str() @@ -74,134 +83,157 @@ class Meta: # Create logical data abstraction (same as data storage for this first example) class PersonSchema(Schema): class Meta: - type_ = 'person' - self_view = 'person_detail' - self_view_kwargs = {'id': ''} - self_view_many = 'person_list' + type_ = "person" + self_view = "person_detail" + self_view_kwargs = {"id": ""} + self_view_many = "person_list" id = fields.Integer(as_string=True, dump_only=True) name = fields.Str(requried=True, load_only=True) email = fields.Email(load_only=True) birth_date = fields.Date() - display_name = fields.Function(lambda obj: "{} <{}>".format(obj.name.upper(), obj.email)) - computers = Relationship(self_view='person_computers', - self_view_kwargs={'id': ''}, - related_view='computer_list', - related_view_kwargs={'id': ''}, - many=True, - schema='ComputerSchema', - type_='computer') + display_name = fields.Function( + lambda obj: "{} <{}>".format(obj.name.upper(), obj.email) + ) + computers = Relationship( + self_view="person_computers", + self_view_kwargs={"id": ""}, + related_view="computer_list", + related_view_kwargs={"id": ""}, + many=True, + schema="ComputerSchema", + type_="computer", + ) tags = fields.Nested(PersonTagSchema, many=True) class ComputerSchema(Schema): class Meta: - type_ = 'computer' - self_view = 'computer_detail' - self_view_kwargs = {'id': ''} + type_ = "computer" + self_view = "computer_detail" + self_view_kwargs = {"id": ""} id = fields.Integer(as_string=True, dump_only=True) serial = fields.Str(requried=True) - owner = Relationship(attribute='person', - self_view='computer_person', - self_view_kwargs={'id': ''}, - related_view='person_detail', - related_view_kwargs={'computer_id': ''}, - schema='PersonSchema', - type_='person') - + owner = Relationship( + attribute="person", + self_view="computer_person", + self_view_kwargs={"id": ""}, + related_view="person_detail", + related_view_kwargs={"computer_id": ""}, + schema="PersonSchema", + type_="person", + ) # Create resource managers class PersonList(ResourceList): schema = PersonSchema - data_layer = {'session': db.session, - 'model': Person} + data_layer = {"session": db.session, "model": Person} class PersonDetail(ResourceDetail): def before_get_object(self, view_kwargs): - if view_kwargs.get('computer_id') is not None: + if view_kwargs.get("computer_id") is not None: try: - computer = self.session.query(Computer).filter_by(id=view_kwargs['computer_id']).one() + computer = ( + self.session.query(Computer) + .filter_by(id=view_kwargs["computer_id"]) + .one() + ) except NoResultFound: - raise ObjectNotFound({'parameter': 'computer_id'}, - "Computer: {} not found".format(view_kwargs['computer_id'])) + raise ObjectNotFound( + {"parameter": "computer_id"}, + "Computer: {} not found".format(view_kwargs["computer_id"]), + ) else: if computer.person is not None: - view_kwargs['id'] = computer.person.id + view_kwargs["id"] = computer.person.id else: - view_kwargs['id'] = None + view_kwargs["id"] = None schema = PersonSchema - data_layer = {'session': db.session, - 'model': Person, - 'methods': {'before_get_object': before_get_object}} + data_layer = { + "session": db.session, + "model": Person, + "methods": {"before_get_object": before_get_object}, + } class PersonRelationship(ResourceRelationship): schema = PersonSchema - data_layer = {'session': db.session, - 'model': Person} + data_layer = {"session": db.session, "model": Person} + class PersonTagList(ResourceList): schema = PersonTagSchema - data_layer = {'session': db.session, - 'model': Person_Tag} + data_layer = {"session": db.session, "model": Person_Tag} + class PersonTagDetail(ResourceDetail): schema = PersonTagSchema - data_layer = {'session': db.session, - 'model': Person_Tag} + data_layer = {"session": db.session, "model": Person_Tag} + class ComputerList(ResourceList): def query(self, view_kwargs): query_ = self.session.query(Computer) - if view_kwargs.get('id') is not None: + if view_kwargs.get("id") is not None: try: - self.session.query(Person).filter_by(id=view_kwargs['id']).one() + self.session.query(Person).filter_by(id=view_kwargs["id"]).one() except NoResultFound: - raise ObjectNotFound({'parameter': 'id'}, "Person: {} not found".format(view_kwargs['id'])) + raise ObjectNotFound( + {"parameter": "id"}, + "Person: {} not found".format(view_kwargs["id"]), + ) else: - query_ = query_.join(Person).filter(Person.id == view_kwargs['id']) + query_ = query_.join(Person).filter(Person.id == view_kwargs["id"]) return query_ def before_create_object(self, data, view_kwargs): - if view_kwargs.get('id') is not None: - person = self.session.query(Person).filter_by(id=view_kwargs['id']).one() - data['person_id'] = person.id + if view_kwargs.get("id") is not None: + person = self.session.query(Person).filter_by(id=view_kwargs["id"]).one() + data["person_id"] = person.id schema = ComputerSchema - data_layer = {'session': db.session, - 'model': Computer, - 'methods': {'query': query, - 'before_create_object': before_create_object}} + data_layer = { + "session": db.session, + "model": Computer, + "methods": {"query": query, "before_create_object": before_create_object}, + } class ComputerDetail(ResourceDetail): schema = ComputerSchema - data_layer = {'session': db.session, - 'model': Computer} + data_layer = {"session": db.session, "model": Computer} class ComputerRelationship(ResourceRelationship): schema = ComputerSchema - data_layer = {'session': db.session, - 'model': Computer} + data_layer = {"session": db.session, "model": Computer} # Create endpoints api = Api(app) -api.route(PersonList, 'person_list', '/persons') -api.route(PersonDetail, 'person_detail', '/persons/', '/computers//owner') -api.route(PersonRelationship, 'person_computers', '/persons//relationships/computers') -api.route(ComputerList, 'computer_list', '/computers', '/persons//computers') -api.route(ComputerDetail, 'computer_detail', '/computers/') -api.route(ComputerRelationship, 'computer_person', '/computers//relationships/owner') - -api.route(PersonTagList, 'person_tag_list', '/persontags') -api.route(PersonTagDetail, 'person_tag_detail', '/persontags/') - -if __name__ == '__main__': +api.route(PersonList, "person_list", "/persons") +api.route( + PersonDetail, + "person_detail", + "/persons/", + "/computers//owner", +) +api.route( + PersonRelationship, "person_computers", "/persons//relationships/computers" +) +api.route(ComputerList, "computer_list", "/computers", "/persons//computers") +api.route(ComputerDetail, "computer_detail", "/computers/") +api.route( + ComputerRelationship, "computer_person", "/computers//relationships/owner" +) + +api.route(PersonTagList, "person_tag_list", "/persontags") +api.route(PersonTagDetail, "person_tag_detail", "/persontags/") + +if __name__ == "__main__": # Start application app.run(debug=True) diff --git a/flask_rest_jsonapi/data_layers/alchemy.py b/flask_rest_jsonapi/data_layers/alchemy.py index 28b8443..ce76793 100644 --- a/flask_rest_jsonapi/data_layers/alchemy.py +++ b/flask_rest_jsonapi/data_layers/alchemy.py @@ -2,20 +2,33 @@ """This module is a CRUD interface between resource managers and the sqlalchemy ORM""" -from sqlalchemy.orm.exc import NoResultFound -from sqlalchemy.orm.collections import InstrumentedList -from sqlalchemy.inspection import inspect -from sqlalchemy.orm.attributes import QueryableAttribute -from sqlalchemy.orm import joinedload, ColumnProperty, RelationshipProperty +from flask import current_app from marshmallow import class_registry from marshmallow.base import SchemaABC +from sqlalchemy.inspection import inspect +from sqlalchemy.orm import ColumnProperty, RelationshipProperty, joinedload +from sqlalchemy.orm.attributes import QueryableAttribute +from sqlalchemy.orm.collections import InstrumentedList +from sqlalchemy.orm.exc import NoResultFound -from flask import current_app from flask_rest_jsonapi.data_layers.base import BaseDataLayer -from flask_rest_jsonapi.exceptions import RelationNotFound, RelatedObjectNotFound, JsonApiException,\ - InvalidSort, ObjectNotFound, InvalidInclude, InvalidType from flask_rest_jsonapi.data_layers.filtering.alchemy import create_filters -from flask_rest_jsonapi.schema import get_model_field, get_related_schema, get_relationships, get_nested_fields, get_schema_field +from flask_rest_jsonapi.exceptions import ( + InvalidInclude, + InvalidSort, + InvalidType, + JsonApiException, + ObjectNotFound, + RelatedObjectNotFound, + RelationNotFound, +) +from flask_rest_jsonapi.schema import ( + get_model_field, + get_nested_fields, + get_related_schema, + get_relationships, + get_schema_field, +) class SqlalchemyDataLayer(BaseDataLayer): @@ -28,12 +41,18 @@ def __init__(self, kwargs): """ super(SqlalchemyDataLayer, self).__init__(kwargs) - if not hasattr(self, 'session'): - raise Exception("You must provide a session in data_layer_kwargs to use sqlalchemy data layer in {}" - .format(self.resource.__name__)) - if not hasattr(self, 'model'): - raise Exception("You must provide a model in data_layer_kwargs to use sqlalchemy data layer in {}" - .format(self.resource.__name__)) + if not hasattr(self, "session"): + raise Exception( + "You must provide a session in data_layer_kwargs to use sqlalchemy data layer in {}".format( + self.resource.__name__ + ) + ) + if not hasattr(self, "model"): + raise Exception( + "You must provide a model in data_layer_kwargs to use sqlalchemy data layer in {}".format( + self.resource.__name__ + ) + ) def create_object(self, data, view_kwargs): """Create an object through sqlalchemy @@ -49,8 +68,9 @@ def create_object(self, data, view_kwargs): join_fields = relationship_fields + nested_fields - obj = self.model(**{key: value - for (key, value) in data.items() if key not in join_fields}) + obj = self.model( + **{key: value for (key, value) in data.items() if key not in join_fields} + ) self.apply_relationships(data, obj) self.apply_nested_fields(data, obj) @@ -62,7 +82,9 @@ def create_object(self, data, view_kwargs): raise e except Exception as e: self.session.rollback() - raise JsonApiException("Object creation error: " + str(e), source={'pointer': '/data'}) + raise JsonApiException( + "Object creation error: " + str(e), source={"pointer": "/data"} + ) self.after_create_object(obj, data, view_kwargs) @@ -76,18 +98,20 @@ def get_object(self, view_kwargs, qs=None): """ self.before_get_object(view_kwargs) - id_field = getattr(self, 'id_field', inspect(self.model).primary_key[0].key) + id_field = getattr(self, "id_field", inspect(self.model).primary_key[0].key) try: filter_field = getattr(self.model, id_field) except Exception: - raise Exception("{} has no attribute {}".format(self.model.__name__, id_field)) + raise Exception( + "{} has no attribute {}".format(self.model.__name__, id_field) + ) - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] query = self.retrieve_object_query(view_kwargs, filter_field, filter_value) - if qs is not None and getattr(self, 'eagerload_includes', True): + if qs is not None and getattr(self, "eagerload_includes", True): query = self.eagerload_includes(query, qs) try: @@ -122,7 +146,7 @@ def get_collection(self, qs, view_kwargs, filters=None): object_count = query.count() - if getattr(self, 'eagerload_includes', True): + if getattr(self, "eagerload_includes", True): query = self.eagerload_includes(query, qs) query = self.paginate_query(query, qs.pagination) @@ -142,10 +166,12 @@ def update_object(self, obj, data, view_kwargs): :return boolean: True if object have changed else False """ if obj is None: - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] - raise ObjectNotFound('{}: {} not found'.format(self.model.__name__, filter_value), - source={'parameter': url_field}) + raise ObjectNotFound( + "{}: {} not found".format(self.model.__name__, filter_value), + source={"parameter": url_field}, + ) self.before_update_object(obj, data, view_kwargs) @@ -168,7 +194,9 @@ def update_object(self, obj, data, view_kwargs): raise e except Exception as e: self.session.rollback() - raise JsonApiException("Update object error: " + str(e), source={'pointer': '/data'}) + raise JsonApiException( + "Update object error: " + str(e), source={"pointer": "/data"} + ) self.after_update_object(obj, data, view_kwargs) @@ -179,10 +207,12 @@ def delete_object(self, obj, view_kwargs): :param dict view_kwargs: kwargs from the resource view """ if obj is None: - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] - raise ObjectNotFound('{}: {} not found'.format(self.model.__name__, filter_value), - source={'parameter': url_field}) + raise ObjectNotFound( + "{}: {} not found".format(self.model.__name__, filter_value), + source={"parameter": url_field}, + ) self.before_delete_object(obj, view_kwargs) @@ -198,7 +228,9 @@ def delete_object(self, obj, view_kwargs): self.after_delete_object(obj, view_kwargs) - def create_relationship(self, json_data, relationship_field, related_id_field, view_kwargs): + def create_relationship( + self, json_data, relationship_field, related_id_field, view_kwargs + ): """Create a relationship :param dict json_data: the request params @@ -207,36 +239,52 @@ def create_relationship(self, json_data, relationship_field, related_id_field, v :param dict view_kwargs: kwargs from the resource view :return boolean: True if relationship have changed else False """ - self.before_create_relationship(json_data, relationship_field, related_id_field, view_kwargs) + self.before_create_relationship( + json_data, relationship_field, related_id_field, view_kwargs + ) obj = self.get_object(view_kwargs) if obj is None: - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] - raise ObjectNotFound('{}: {} not found'.format(self.model.__name__, filter_value), - source={'parameter': url_field}) + raise ObjectNotFound( + "{}: {} not found".format(self.model.__name__, filter_value), + source={"parameter": url_field}, + ) if not hasattr(obj, relationship_field): - raise RelationNotFound("{} has no attribute {}".format(obj.__class__.__name__, relationship_field)) + raise RelationNotFound( + "{} has no attribute {}".format( + obj.__class__.__name__, relationship_field + ) + ) - related_model = getattr(obj.__class__, relationship_field).property.mapper.class_ + related_model = getattr( + obj.__class__, relationship_field + ).property.mapper.class_ updated = False - if isinstance(json_data['data'], list): - obj_ids = {str(getattr(obj__, related_id_field)) for obj__ in getattr(obj, relationship_field)} - - for obj_ in json_data['data']: - if obj_['id'] not in obj_ids: - getattr(obj, - relationship_field).append(self.get_related_object(related_model, related_id_field, obj_)) + if isinstance(json_data["data"], list): + obj_ids = { + str(getattr(obj__, related_id_field)) + for obj__ in getattr(obj, relationship_field) + } + + for obj_ in json_data["data"]: + if obj_["id"] not in obj_ids: + getattr(obj, relationship_field).append( + self.get_related_object(related_model, related_id_field, obj_) + ) updated = True else: related_object = None - if json_data['data'] is not None: - related_object = self.get_related_object(related_model, related_id_field, json_data['data']) + if json_data["data"] is not None: + related_object = self.get_related_object( + related_model, related_id_field, json_data["data"] + ) obj_id = getattr(getattr(obj, relationship_field), related_id_field, None) new_obj_id = getattr(related_object, related_id_field, None) @@ -253,11 +301,15 @@ def create_relationship(self, json_data, relationship_field, related_id_field, v self.session.rollback() raise JsonApiException("Create relationship error: " + str(e)) - self.after_create_relationship(obj, updated, json_data, relationship_field, related_id_field, view_kwargs) + self.after_create_relationship( + obj, updated, json_data, relationship_field, related_id_field, view_kwargs + ) return obj, updated - def get_relationship(self, relationship_field, related_type_, related_id_field, view_kwargs): + def get_relationship( + self, relationship_field, related_type_, related_id_field, view_kwargs + ): """Get a relationship :param str relationship_field: the model attribute used for relationship @@ -266,34 +318,55 @@ def get_relationship(self, relationship_field, related_type_, related_id_field, :param dict view_kwargs: kwargs from the resource view :return tuple: the object and related object(s) """ - self.before_get_relationship(relationship_field, related_type_, related_id_field, view_kwargs) + self.before_get_relationship( + relationship_field, related_type_, related_id_field, view_kwargs + ) obj = self.get_object(view_kwargs) if obj is None: - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] - raise ObjectNotFound('{}: {} not found'.format(self.model.__name__, filter_value), - source={'parameter': url_field}) + raise ObjectNotFound( + "{}: {} not found".format(self.model.__name__, filter_value), + source={"parameter": url_field}, + ) if not hasattr(obj, relationship_field): - raise RelationNotFound("{} has no attribute {}".format(obj.__class__.__name__, relationship_field)) + raise RelationNotFound( + "{} has no attribute {}".format( + obj.__class__.__name__, relationship_field + ) + ) related_objects = getattr(obj, relationship_field) if related_objects is None: return obj, related_objects - self.after_get_relationship(obj, related_objects, relationship_field, related_type_, related_id_field, - view_kwargs) + self.after_get_relationship( + obj, + related_objects, + relationship_field, + related_type_, + related_id_field, + view_kwargs, + ) if isinstance(related_objects, InstrumentedList): - return obj,\ - [{'type': related_type_, 'id': getattr(obj_, related_id_field)} for obj_ in related_objects] + return obj, [ + {"type": related_type_, "id": getattr(obj_, related_id_field)} + for obj_ in related_objects + ] else: - return obj, {'type': related_type_, 'id': getattr(related_objects, related_id_field)} - - def update_relationship(self, json_data, relationship_field, related_id_field, view_kwargs): + return obj, { + "type": related_type_, + "id": getattr(related_objects, related_id_field), + } + + def update_relationship( + self, json_data, relationship_field, related_id_field, view_kwargs + ): """Update a relationship :param dict json_data: the request params @@ -302,31 +375,49 @@ def update_relationship(self, json_data, relationship_field, related_id_field, v :param dict view_kwargs: kwargs from the resource view :return boolean: True if relationship have changed else False """ - self.before_update_relationship(json_data, relationship_field, related_id_field, view_kwargs) + self.before_update_relationship( + json_data, relationship_field, related_id_field, view_kwargs + ) obj = self.get_object(view_kwargs) if obj is None: - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] - raise ObjectNotFound('{}: {} not found'.format(self.model.__name__, filter_value), - source={'parameter': url_field}) + raise ObjectNotFound( + "{}: {} not found".format(self.model.__name__, filter_value), + source={"parameter": url_field}, + ) if not hasattr(obj, relationship_field): - raise RelationNotFound("{} has no attribute {}".format(obj.__class__.__name__, relationship_field)) + raise RelationNotFound( + "{} has no attribute {}".format( + obj.__class__.__name__, relationship_field + ) + ) - related_model = getattr(obj.__class__, relationship_field).property.mapper.class_ + related_model = getattr( + obj.__class__, relationship_field + ).property.mapper.class_ updated = False - if isinstance(json_data['data'], list): + if isinstance(json_data["data"], list): related_objects = [] - for obj_ in json_data['data']: - related_objects.append(self.get_related_object(related_model, related_id_field, obj_)) - - obj_ids = {getattr(obj__, related_id_field) for obj__ in getattr(obj, relationship_field)} - new_obj_ids = {getattr(related_object, related_id_field) for related_object in related_objects} + for obj_ in json_data["data"]: + related_objects.append( + self.get_related_object(related_model, related_id_field, obj_) + ) + + obj_ids = { + getattr(obj__, related_id_field) + for obj__ in getattr(obj, relationship_field) + } + new_obj_ids = { + getattr(related_object, related_id_field) + for related_object in related_objects + } if obj_ids != new_obj_ids: setattr(obj, relationship_field, related_objects) updated = True @@ -334,8 +425,10 @@ def update_relationship(self, json_data, relationship_field, related_id_field, v else: related_object = None - if json_data['data'] is not None: - related_object = self.get_related_object(related_model, related_id_field, json_data['data']) + if json_data["data"] is not None: + related_object = self.get_related_object( + related_model, related_id_field, json_data["data"] + ) obj_id = getattr(getattr(obj, relationship_field), related_id_field, None) new_obj_id = getattr(related_object, related_id_field, None) @@ -352,11 +445,15 @@ def update_relationship(self, json_data, relationship_field, related_id_field, v self.session.rollback() raise JsonApiException("Update relationship error: " + str(e)) - self.after_update_relationship(obj, updated, json_data, relationship_field, related_id_field, view_kwargs) + self.after_update_relationship( + obj, updated, json_data, relationship_field, related_id_field, view_kwargs + ) return obj, updated - def delete_relationship(self, json_data, relationship_field, related_id_field, view_kwargs): + def delete_relationship( + self, json_data, relationship_field, related_id_field, view_kwargs + ): """Delete a relationship :param dict json_data: the request params @@ -364,30 +461,44 @@ def delete_relationship(self, json_data, relationship_field, related_id_field, v :param str related_id_field: the identifier field of the related model :param dict view_kwargs: kwargs from the resource view """ - self.before_delete_relationship(json_data, relationship_field, related_id_field, view_kwargs) + self.before_delete_relationship( + json_data, relationship_field, related_id_field, view_kwargs + ) obj = self.get_object(view_kwargs) if obj is None: - url_field = getattr(self, 'url_field', 'id') + url_field = getattr(self, "url_field", "id") filter_value = view_kwargs[url_field] - raise ObjectNotFound('{}: {} not found'.format(self.model.__name__, filter_value), - source={'parameter': url_field}) + raise ObjectNotFound( + "{}: {} not found".format(self.model.__name__, filter_value), + source={"parameter": url_field}, + ) if not hasattr(obj, relationship_field): - raise RelationNotFound("{} has no attribute {}".format(obj.__class__.__name__, relationship_field)) + raise RelationNotFound( + "{} has no attribute {}".format( + obj.__class__.__name__, relationship_field + ) + ) - related_model = getattr(obj.__class__, relationship_field).property.mapper.class_ + related_model = getattr( + obj.__class__, relationship_field + ).property.mapper.class_ updated = False - if isinstance(json_data['data'], list): - obj_ids = {str(getattr(obj__, related_id_field)) for obj__ in getattr(obj, relationship_field)} - - for obj_ in json_data['data']: - if obj_['id'] in obj_ids: - getattr(obj, - relationship_field).remove(self.get_related_object(related_model, related_id_field, obj_)) + if isinstance(json_data["data"], list): + obj_ids = { + str(getattr(obj__, related_id_field)) + for obj__ in getattr(obj, relationship_field) + } + + for obj_ in json_data["data"]: + if obj_["id"] in obj_ids: + getattr(obj, relationship_field).remove( + self.get_related_object(related_model, related_id_field, obj_) + ) updated = True else: setattr(obj, relationship_field, None) @@ -402,7 +513,9 @@ def delete_relationship(self, json_data, relationship_field, related_id_field, v self.session.rollback() raise JsonApiException("Delete relationship error: " + str(e)) - self.after_delete_relationship(obj, updated, json_data, relationship_field, related_id_field, view_kwargs) + self.after_delete_relationship( + obj, updated, json_data, relationship_field, related_id_field, view_kwargs + ) return obj, updated @@ -415,17 +528,20 @@ def get_related_object(self, related_model, related_id_field, obj): :return DeclarativeMeta: a related object """ try: - related_object = self.session.query(related_model)\ - .filter(getattr(related_model, related_id_field) == obj['id'])\ - .one() + related_object = ( + self.session.query(related_model) + .filter(getattr(related_model, related_id_field) == obj["id"]) + .one() + ) except NoResultFound: - raise RelatedObjectNotFound("{}.{}: {} not found".format(related_model.__name__, - related_id_field, - obj['id'])) + raise RelatedObjectNotFound( + "{}.{}: {} not found".format( + related_model.__name__, related_id_field, obj["id"] + ) + ) return related_object - def apply_relationships(self, data, obj): """Apply relationship provided by data to obj @@ -439,26 +555,36 @@ def apply_relationships(self, data, obj): if key in relationship_fields: related_model = getattr(obj.__class__, key).property.mapper.class_ schema_field = get_schema_field(self.resource.schema, key) - related_id_field = self.resource.schema._declared_fields[schema_field].id_field + related_id_field = self.resource.schema._declared_fields[ + schema_field + ].id_field if isinstance(value, list): related_objects = [] for identifier in value: - related_object = self.get_related_object(related_model, related_id_field, {'id': identifier}) + related_object = self.get_related_object( + related_model, related_id_field, {"id": identifier} + ) related_objects.append(related_object) - relationships_to_apply.append({'field': key, 'value': related_objects}) + relationships_to_apply.append( + {"field": key, "value": related_objects} + ) else: related_object = None if value is not None: - related_object = self.get_related_object(related_model, related_id_field, {'id': value}) + related_object = self.get_related_object( + related_model, related_id_field, {"id": value} + ) - relationships_to_apply.append({'field': key, 'value': related_object}) + relationships_to_apply.append( + {"field": key, "value": related_object} + ) for relationship in relationships_to_apply: - setattr(obj, relationship['field'], relationship['value']) + setattr(obj, relationship["field"], relationship["value"]) def apply_nested_fields(self, data, obj): nested_fields_to_apply = [] @@ -468,7 +594,9 @@ def apply_nested_fields(self, data, obj): nested_field_inspection = inspect(getattr(obj.__class__, key)) if not isinstance(nested_field_inspection, QueryableAttribute): - raise InvalidType("Unrecognized nested field type: not a queryable attribute.") + raise InvalidType( + "Unrecognized nested field type: not a queryable attribute." + ) if isinstance(nested_field_inspection.property, RelationshipProperty): nested_model = getattr(obj.__class__, key).property.mapper.class_ @@ -480,16 +608,22 @@ def apply_nested_fields(self, data, obj): nested_object = nested_model(**identifier) nested_objects.append(nested_object) - nested_fields_to_apply.append({'field': key, 'value': nested_objects}) + nested_fields_to_apply.append( + {"field": key, "value": nested_objects} + ) else: - nested_fields_to_apply.append({'field': key, 'value': nested_model(**value)}) + nested_fields_to_apply.append( + {"field": key, "value": nested_model(**value)} + ) elif isinstance(nested_field_inspection.property, ColumnProperty): - nested_fields_to_apply.append({'field': key, 'value': value}) + nested_fields_to_apply.append({"field": key, "value": value}) else: - raise InvalidType("Unrecognized nested field type: not a RelationshipProperty or ColumnProperty.") + raise InvalidType( + "Unrecognized nested field type: not a RelationshipProperty or ColumnProperty." + ) for nested_field in nested_fields_to_apply: - setattr(obj, nested_field['field'], nested_field['value']) + setattr(obj, nested_field["field"], nested_field["value"]) def filter_query(self, query, filter_info, model): """Filter query according to jsonapi 1.0 @@ -514,10 +648,14 @@ def sort_query(self, query, sort_info): :return Query: the sorted query """ for sort_opt in sort_info: - field = sort_opt['field'] + field = sort_opt["field"] if not hasattr(self.model, field): - raise InvalidSort("{} has no attribute {}".format(self.model.__name__, field)) - query = query.order_by(getattr(getattr(self.model, field), sort_opt['order'])()) + raise InvalidSort( + "{} has no attribute {}".format(self.model.__name__, field) + ) + query = query.order_by( + getattr(getattr(self.model, field), sort_opt["order"])() + ) return query def paginate_query(self, query, paginate_info): @@ -527,13 +665,13 @@ def paginate_query(self, query, paginate_info): :param dict paginate_info: pagination information :return Query: the paginated query """ - if int(paginate_info.get('size', 1)) == 0: + if int(paginate_info.get("size", 1)) == 0: return query - page_size = int(paginate_info.get('size', 0)) or current_app.config['PAGE_SIZE'] + page_size = int(paginate_info.get("size", 0)) or current_app.config["PAGE_SIZE"] query = query.limit(page_size) - if paginate_info.get('number'): - query = query.offset((int(paginate_info['number']) - 1) * page_size) + if paginate_info.get("number"): + query = query.offset((int(paginate_info["number"]) - 1) * page_size) return query @@ -547,30 +685,37 @@ def eagerload_includes(self, query, qs): for include in qs.include: joinload_object = None - if '.' in include: + if "." in include: current_schema = self.resource.schema - for obj in include.split('.'): + current_model = self.model + for obj in include.split("."): try: field = get_model_field(current_schema, obj) except Exception as e: raise InvalidInclude(str(e)) if joinload_object is None: - joinload_object = joinedload(field) + joinload_object = joinedload(getattr(current_model, field)) else: - joinload_object = joinload_object.joinedload(field) + joinload_object = joinload_object.joinedload( + getattr(current_model, field) + ) related_schema_cls = get_related_schema(current_schema, obj) if isinstance(related_schema_cls, SchemaABC): related_schema_cls = related_schema_cls.__class__ else: - related_schema_cls = class_registry.get_class(related_schema_cls) + related_schema_cls = class_registry.get_class( + related_schema_cls + ) current_schema = related_schema_cls + current_model = getattr(current_model, field).mapper.class_ else: try: - field = get_model_field(self.resource.schema, include) + current_model = self.model + field = getattr(current_model, include) except Exception as e: raise InvalidInclude(str(e)) @@ -680,7 +825,9 @@ def after_delete_object(self, obj, view_kwargs): """ pass - def before_create_relationship(self, json_data, relationship_field, related_id_field, view_kwargs): + def before_create_relationship( + self, json_data, relationship_field, related_id_field, view_kwargs + ): """Make work before to create a relationship :param dict json_data: the request params @@ -691,7 +838,9 @@ def before_create_relationship(self, json_data, relationship_field, related_id_f """ pass - def after_create_relationship(self, obj, updated, json_data, relationship_field, related_id_field, view_kwargs): + def after_create_relationship( + self, obj, updated, json_data, relationship_field, related_id_field, view_kwargs + ): """Make work after to create a relationship :param obj: an object from data layer @@ -704,7 +853,9 @@ def after_create_relationship(self, obj, updated, json_data, relationship_field, """ pass - def before_get_relationship(self, relationship_field, related_type_, related_id_field, view_kwargs): + def before_get_relationship( + self, relationship_field, related_type_, related_id_field, view_kwargs + ): """Make work before to get information about a relationship :param str relationship_field: the model attribute used for relationship @@ -715,8 +866,15 @@ def before_get_relationship(self, relationship_field, related_type_, related_id_ """ pass - def after_get_relationship(self, obj, related_objects, relationship_field, related_type_, related_id_field, - view_kwargs): + def after_get_relationship( + self, + obj, + related_objects, + relationship_field, + related_type_, + related_id_field, + view_kwargs, + ): """Make work after to get information about a relationship :param obj: an object from data layer @@ -729,7 +887,9 @@ def after_get_relationship(self, obj, related_objects, relationship_field, relat """ pass - def before_update_relationship(self, json_data, relationship_field, related_id_field, view_kwargs): + def before_update_relationship( + self, json_data, relationship_field, related_id_field, view_kwargs + ): """Make work before to update a relationship :param dict json_data: the request params @@ -740,7 +900,9 @@ def before_update_relationship(self, json_data, relationship_field, related_id_f """ pass - def after_update_relationship(self, obj, updated, json_data, relationship_field, related_id_field, view_kwargs): + def after_update_relationship( + self, obj, updated, json_data, relationship_field, related_id_field, view_kwargs + ): """Make work after to update a relationship :param obj: an object from data layer @@ -753,7 +915,9 @@ def after_update_relationship(self, obj, updated, json_data, relationship_field, """ pass - def before_delete_relationship(self, json_data, relationship_field, related_id_field, view_kwargs): + def before_delete_relationship( + self, json_data, relationship_field, related_id_field, view_kwargs + ): """Make work before to delete a relationship :param dict json_data: the request params @@ -763,7 +927,9 @@ def before_delete_relationship(self, json_data, relationship_field, related_id_f """ pass - def after_delete_relationship(self, obj, updated, json_data, relationship_field, related_id_field, view_kwargs): + def after_delete_relationship( + self, obj, updated, json_data, relationship_field, related_id_field, view_kwargs + ): """Make work after to delete a relationship :param obj: an object from data layer diff --git a/flask_rest_jsonapi/resource.py b/flask_rest_jsonapi/resource.py index 26b9318..2102bf5 100644 --- a/flask_rest_jsonapi/resource.py +++ b/flask_rest_jsonapi/resource.py @@ -4,24 +4,58 @@ import inspect import json -from six import with_metaclass -from werkzeug.wrappers import Response -from flask import request, url_for, make_response +from flask import make_response, request, url_for +from flask.views import MethodView from flask.wrappers import Response as FlaskResponse -from flask.views import MethodView, MethodViewType -from marshmallow_jsonapi.exceptions import IncorrectTypeError from marshmallow import ValidationError +from marshmallow_jsonapi.exceptions import IncorrectTypeError +from marshmallow_jsonapi.fields import BaseRelationship +from werkzeug.wrappers import Response -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, RelationNotFound -from flask_rest_jsonapi.decorators import check_headers, check_method_requirements, jsonapi_exception_formatter -from flask_rest_jsonapi.schema import compute_schema, get_relationships, get_model_field -from flask_rest_jsonapi.data_layers.base import BaseDataLayer from flask_rest_jsonapi.data_layers.alchemy import SqlalchemyDataLayer +from flask_rest_jsonapi.data_layers.base import BaseDataLayer +from flask_rest_jsonapi.decorators import ( + check_headers, + check_method_requirements, + jsonapi_exception_formatter, +) +from flask_rest_jsonapi.exceptions import BadRequest, InvalidType, RelationNotFound +from flask_rest_jsonapi.pagination import add_pagination_links +from flask_rest_jsonapi.querystring import QueryStringManager as QSManager +from flask_rest_jsonapi.schema import compute_schema, get_model_field, get_relationships from flask_rest_jsonapi.utils import JSONEncoder -from marshmallow_jsonapi.fields import BaseRelationship + +http_method_funcs = frozenset( + ["get", "post", "head", "options", "delete", "put", "trace", "patch"] +) + + +class MethodViewType(type): + """Metaclass for :class:`MethodView` that determines what methods the view + defines. + """ + + def __init__(cls, name, bases, d): + super().__init__(name, bases, d) + + if "methods" not in d: + methods = set() + + for base in bases: + if getattr(base, "methods", None): + methods.update(base.methods) + + for key in http_method_funcs: + if hasattr(cls, key): + methods.add(key.upper()) + + # If we have no method at all in there we don't want to add a + # method list. This is for instance the case for the base class + # or another subclass of a base method view that does not introduce + # new methods. + if methods: + cls.methods = methods class ResourceMeta(MethodViewType): @@ -30,22 +64,32 @@ class ResourceMeta(MethodViewType): def __new__(cls, name, bases, d): """Constructor of a resource class""" rv = super(ResourceMeta, cls).__new__(cls, name, bases, d) - if 'data_layer' in d: - if not isinstance(d['data_layer'], dict): - raise Exception("You must provide a data layer information as dict in {}".format(cls.__name__)) - - if d['data_layer'].get('class') is not None\ - and BaseDataLayer not in inspect.getmro(d['data_layer']['class']): - raise Exception("You must provide a data layer class inherited from BaseDataLayer in {}" - .format(cls.__name__)) - - data_layer_cls = d['data_layer'].get('class', SqlalchemyDataLayer) - data_layer_kwargs = d['data_layer'] + if "data_layer" in d: + if not isinstance(d["data_layer"], dict): + raise Exception( + "You must provide a data layer information as dict in {}".format( + cls.__name__ + ) + ) + + if d["data_layer"].get( + "class" + ) is not None and BaseDataLayer not in inspect.getmro( + d["data_layer"]["class"] + ): + raise Exception( + "You must provide a data layer class inherited from BaseDataLayer in {}".format( + cls.__name__ + ) + ) + + data_layer_cls = d["data_layer"].get("class", SqlalchemyDataLayer) + data_layer_kwargs = d["data_layer"] rv._data_layer = data_layer_cls(data_layer_kwargs) rv.decorators = (check_headers,) - if 'decorators' in d: - rv.decorators += d['decorators'] + if "decorators" in d: + rv.decorators += d["decorators"] return rv @@ -55,7 +99,7 @@ class Resource(MethodView): def __new__(cls, *args, **kwargs): """Constructor of a resource instance""" - if hasattr(cls, '_data_layer'): + if hasattr(cls, "_data_layer"): cls._data_layer.resource = cls return super(Resource, cls).__new__(cls) @@ -64,26 +108,26 @@ def __new__(cls, *args, **kwargs): def dispatch_request(self, *args, **kwargs): """Logic of how to handle a request""" 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) + 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'} + headers = {"Content-Type": "application/vnd.api+json"} response = method(*args, **kwargs) if isinstance(response, Response): - response.headers.add('Content-Type', 'application/vnd.api+json') + 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'}}) + response.update({"jsonapi": {"version": "1.0"}}) return make_response(json.dumps(response, cls=JSONEncoder), 200, headers) try: data, status_code, headers = response - headers.update({'Content-Type': 'application/vnd.api+json'}) + headers.update({"Content-Type": "application/vnd.api+json"}) except ValueError: pass @@ -93,10 +137,10 @@ def dispatch_request(self, *args, **kwargs): pass if isinstance(data, dict): - data.update({'jsonapi': {'version': '1.0'}}) + data.update({"jsonapi": {"version": "1.0"}}) if isinstance(data, FlaskResponse): - data.headers.add('Content-Type', 'application/vnd.api+json') + data.headers.add("Content-Type", "application/vnd.api+json") data.status_code = status_code return data elif isinstance(data, str): @@ -107,7 +151,7 @@ def dispatch_request(self, *args, **kwargs): return make_response(json_reponse, status_code, headers) -class ResourceList(with_metaclass(ResourceMeta, Resource)): +class ResourceList(Resource, metaclass=ResourceMeta): """Base class of a resource list manager""" @check_method_requirements @@ -120,25 +164,23 @@ def get(self, *args, **kwargs): parent_filter = self._get_parent_filter(request.url, kwargs) objects_count, objects = self.get_collection(qs, kwargs, filters=parent_filter) - schema_kwargs = getattr(self, 'get_schema_kwargs', dict()) - schema_kwargs.update({'many': True}) + schema_kwargs = getattr(self, "get_schema_kwargs", dict()) + schema_kwargs.update({"many": True}) self.before_marshmallow(args, kwargs) - schema = compute_schema(self.schema, - schema_kwargs, - qs, - qs.include) + schema = compute_schema(self.schema, schema_kwargs, qs, qs.include) result = schema.dump(objects) - view_kwargs = request.view_args if getattr(self, 'view_kwargs', None) is True else dict() - add_pagination_links(result, - objects_count, - qs, - url_for(self.view, _external=True, **view_kwargs)) + view_kwargs = ( + request.view_args if getattr(self, "view_kwargs", None) is True else dict() + ) + add_pagination_links( + result, objects_count, qs, url_for(self.view, _external=True, **view_kwargs) + ) - result.update({'meta': {'count': objects_count}}) + result.update({"meta": {"count": objects_count}}) final_result = self.after_get(result) @@ -153,24 +195,23 @@ def post(self, *args, **kwargs): self.before_marshmallow(args, kwargs) - schema = compute_schema(self.schema, - getattr(self, 'post_schema_kwargs', dict()), - qs, - qs.include) + schema = compute_schema( + self.schema, getattr(self, "post_schema_kwargs", dict()), qs, qs.include + ) try: data = schema.load(json_data) except IncorrectTypeError as e: errors = e.messages - for error in errors['errors']: - error['status'] = '409' - error['title'] = "Incorrect type" + 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" + for message in errors["errors"]: + message["status"] = "422" + message["title"] = "Validation error" return errors, 422 self.before_post(args, kwargs, data=data) @@ -179,8 +220,8 @@ def post(self, *args, **kwargs): result = schema.dump(obj) - if result['data'].get('links', {}).get('self'): - final_result = (result, 201, {'Location': result['data']['links']['self']}) + if result["data"].get("links", {}).get("self"): + final_result = (result, 201, {"Location": result["data"]["links"]["self"]}) else: final_result = (result, 201) @@ -194,7 +235,7 @@ def _get_parent_filter(self, url, kwargs): belonging to the parent resource are returned """ - url_segments = url.split('/') + url_segments = url.split("/") parent_segment = url_segments[-3] parent_id = url_segments[-2] @@ -231,7 +272,7 @@ def create_object(self, data, kwargs): return self._data_layer.create_object(data, kwargs) -class ResourceDetail(with_metaclass(ResourceMeta, Resource)): +class ResourceDetail(Resource, metaclass=ResourceMeta): """Base class of a resource detail manager""" @check_method_requirements @@ -245,10 +286,9 @@ def get(self, *args, **kwargs): self.before_marshmallow(args, kwargs) - schema = compute_schema(self.schema, - getattr(self, 'get_schema_kwargs', dict()), - qs, - qs.include) + schema = compute_schema( + self.schema, getattr(self, "get_schema_kwargs", dict()), qs, qs.include + ) result = schema.dump(obj) if obj else None @@ -262,36 +302,38 @@ def patch(self, *args, **kwargs): json_data = request.get_json() or {} qs = QSManager(request.args, self.schema) - schema_kwargs = getattr(self, 'patch_schema_kwargs', dict()) + schema_kwargs = getattr(self, "patch_schema_kwargs", dict()) self.before_marshmallow(args, kwargs) - schema = compute_schema(self.schema, - schema_kwargs, - qs, - qs.include) + schema = compute_schema(self.schema, schema_kwargs, qs, qs.include) try: data = schema.load(json_data, partial=True) except IncorrectTypeError as e: errors = e.messages - for error in errors['errors']: - error['status'] = '409' - error['title'] = "Incorrect type" + 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" + for message in errors["errors"]: + message["status"] = "422" + message["title"] = "Validation error" return errors, 422 - if 'id' not in json_data['data']: - raise BadRequest('Missing id in "data" node', - source={'pointer': '/data/id'}) - if (str(json_data['data']['id']) != str(kwargs[getattr(self._data_layer, 'url_field', 'id')])): - raise BadRequest('Value of id does not match the resource identifier in url', - source={'pointer': '/data/id'}) + if "id" not in json_data["data"]: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if str(json_data["data"]["id"]) != str( + kwargs[getattr(self._data_layer, "url_field", "id")] + ): + raise BadRequest( + "Value of id does not match the resource identifier in url", + source={"pointer": "/data/id"}, + ) self.before_patch(args, kwargs, data=data) @@ -310,7 +352,7 @@ def delete(self, *args, **kwargs): self.delete_object(kwargs) - result = {'meta': {'message': 'Object successfully deleted'}} + result = {"meta": {"message": "Object successfully deleted"}} final_result = self.after_delete(result) @@ -357,7 +399,7 @@ def delete_object(self, kwargs): self._data_layer.delete_object(obj, kwargs) -class ResourceRelationship(with_metaclass(ResourceMeta, Resource)): +class ResourceRelationship(Resource, metaclass=ResourceMeta): """Base class of a resource relationship manager""" @check_method_requirements @@ -365,23 +407,33 @@ 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() - - obj, data = self._data_layer.get_relationship(model_relationship_field, - related_type_, - related_id_field, - kwargs) - - result = {'links': {'self': request.path, - 'related': self.schema._declared_fields[relationship_field].get_related_url(obj)}, - 'data': data} + ( + relationship_field, + model_relationship_field, + related_type_, + related_id_field, + ) = self._get_relationship_data() + + obj, data = self._data_layer.get_relationship( + model_relationship_field, related_type_, related_id_field, kwargs + ) + + result = { + "links": { + "self": request.path, + "related": self.schema._declared_fields[ + relationship_field + ].get_related_url(obj), + }, + "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.get('included', dict()) + result["included"] = serialized_obj.get("included", dict()) final_result = self.after_get(result) @@ -392,39 +444,59 @@ def post(self, *args, **kwargs): """Add / create relationship(s)""" json_data = request.get_json() or {} - relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() - - if 'data' not in json_data: - raise BadRequest('You must provide data with a "data" route node', source={'pointer': '/data'}) - if isinstance(json_data['data'], dict): - if 'type' not in json_data['data']: - raise BadRequest('Missing type in "data" node', source={'pointer': '/data/type'}) - if 'id' not in json_data['data']: - raise BadRequest('Missing id in "data" node', source={'pointer': '/data/id'}) - if json_data['data']['type'] != related_type_: - raise InvalidType('The type field does not match the resource type', source={'pointer': '/data/type'}) - if isinstance(json_data['data'], list): - for obj in json_data['data']: - if 'type' not in obj: - raise BadRequest('Missing type in "data" node', source={'pointer': '/data/type'}) - if 'id' not in obj: - raise BadRequest('Missing id in "data" node', source={'pointer': '/data/id'}) - if obj['type'] != related_type_: - raise InvalidType('The type provided does not match the resource type', - source={'pointer': '/data/type'}) + ( + relationship_field, + model_relationship_field, + related_type_, + related_id_field, + ) = self._get_relationship_data() + + if "data" not in json_data: + raise BadRequest( + 'You must provide data with a "data" route node', + source={"pointer": "/data"}, + ) + if isinstance(json_data["data"], dict): + if "type" not in json_data["data"]: + raise BadRequest( + 'Missing type in "data" node', source={"pointer": "/data/type"} + ) + if "id" not in json_data["data"]: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if json_data["data"]["type"] != related_type_: + raise InvalidType( + "The type field does not match the resource type", + source={"pointer": "/data/type"}, + ) + if isinstance(json_data["data"], list): + for obj in json_data["data"]: + if "type" not in obj: + raise BadRequest( + 'Missing type in "data" node', source={"pointer": "/data/type"} + ) + if "id" not in obj: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if obj["type"] != related_type_: + raise InvalidType( + "The type provided does not match the resource type", + source={"pointer": "/data/type"}, + ) self.before_post(args, kwargs, json_data=json_data) - obj_, updated = self._data_layer.create_relationship(json_data, - model_relationship_field, - related_id_field, - kwargs) + obj_, updated = self._data_layer.create_relationship( + json_data, model_relationship_field, related_id_field, kwargs + ) status_code = 200 - result = {'meta': {'message': 'Relationship successfully created'}} + result = {"meta": {"message": "Relationship successfully created"}} if updated is False: - result = '' + result = "" status_code = 204 final_result = self.after_post(result, status_code) @@ -436,39 +508,59 @@ def patch(self, *args, **kwargs): """Update a relationship""" json_data = request.get_json() or {} - relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() - - if 'data' not in json_data: - raise BadRequest('You must provide data with a "data" route node', source={'pointer': '/data'}) - if isinstance(json_data['data'], dict): - if 'type' not in json_data['data']: - raise BadRequest('Missing type in "data" node', source={'pointer': '/data/type'}) - if 'id' not in json_data['data']: - raise BadRequest('Missing id in "data" node', source={'pointer': '/data/id'}) - if json_data['data']['type'] != related_type_: - raise InvalidType('The type field does not match the resource type', source={'pointer': '/data/type'}) - if isinstance(json_data['data'], list): - for obj in json_data['data']: - if 'type' not in obj: - raise BadRequest('Missing type in "data" node', source={'pointer': '/data/type'}) - if 'id' not in obj: - raise BadRequest('Missing id in "data" node', source={'pointer': '/data/id'}) - if obj['type'] != related_type_: - raise InvalidType('The type provided does not match the resource type', - source={'pointer': '/data/type'}) + ( + relationship_field, + model_relationship_field, + related_type_, + related_id_field, + ) = self._get_relationship_data() + + if "data" not in json_data: + raise BadRequest( + 'You must provide data with a "data" route node', + source={"pointer": "/data"}, + ) + if isinstance(json_data["data"], dict): + if "type" not in json_data["data"]: + raise BadRequest( + 'Missing type in "data" node', source={"pointer": "/data/type"} + ) + if "id" not in json_data["data"]: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if json_data["data"]["type"] != related_type_: + raise InvalidType( + "The type field does not match the resource type", + source={"pointer": "/data/type"}, + ) + if isinstance(json_data["data"], list): + for obj in json_data["data"]: + if "type" not in obj: + raise BadRequest( + 'Missing type in "data" node', source={"pointer": "/data/type"} + ) + if "id" not in obj: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if obj["type"] != related_type_: + raise InvalidType( + "The type provided does not match the resource type", + source={"pointer": "/data/type"}, + ) self.before_patch(args, kwargs, json_data=json_data) - obj_, updated = self._data_layer.update_relationship(json_data, - model_relationship_field, - related_id_field, - kwargs) + obj_, updated = self._data_layer.update_relationship( + json_data, model_relationship_field, related_id_field, kwargs + ) status_code = 200 - result = {'meta': {'message': 'Relationship successfully updated'}} + result = {"meta": {"message": "Relationship successfully updated"}} if updated is False: - result = '' + result = "" status_code = 204 final_result = self.after_patch(result, status_code) @@ -480,39 +572,59 @@ def delete(self, *args, **kwargs): """Delete relationship(s)""" json_data = request.get_json() or {} - relationship_field, model_relationship_field, related_type_, related_id_field = self._get_relationship_data() - - if 'data' not in json_data: - raise BadRequest('You must provide data with a "data" route node', source={'pointer': '/data'}) - if isinstance(json_data['data'], dict): - if 'type' not in json_data['data']: - raise BadRequest('Missing type in "data" node', source={'pointer': '/data/type'}) - if 'id' not in json_data['data']: - raise BadRequest('Missing id in "data" node', source={'pointer': '/data/id'}) - if json_data['data']['type'] != related_type_: - raise InvalidType('The type field does not match the resource type', source={'pointer': '/data/type'}) - if isinstance(json_data['data'], list): - for obj in json_data['data']: - if 'type' not in obj: - raise BadRequest('Missing type in "data" node', source={'pointer': '/data/type'}) - if 'id' not in obj: - raise BadRequest('Missing id in "data" node', source={'pointer': '/data/id'}) - if obj['type'] != related_type_: - raise InvalidType('The type provided does not match the resource type', - source={'pointer': '/data/type'}) + ( + relationship_field, + model_relationship_field, + related_type_, + related_id_field, + ) = self._get_relationship_data() + + if "data" not in json_data: + raise BadRequest( + 'You must provide data with a "data" route node', + source={"pointer": "/data"}, + ) + if isinstance(json_data["data"], dict): + if "type" not in json_data["data"]: + raise BadRequest( + 'Missing type in "data" node', source={"pointer": "/data/type"} + ) + if "id" not in json_data["data"]: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if json_data["data"]["type"] != related_type_: + raise InvalidType( + "The type field does not match the resource type", + source={"pointer": "/data/type"}, + ) + if isinstance(json_data["data"], list): + for obj in json_data["data"]: + if "type" not in obj: + raise BadRequest( + 'Missing type in "data" node', source={"pointer": "/data/type"} + ) + if "id" not in obj: + raise BadRequest( + 'Missing id in "data" node', source={"pointer": "/data/id"} + ) + if obj["type"] != related_type_: + raise InvalidType( + "The type provided does not match the resource type", + source={"pointer": "/data/type"}, + ) self.before_delete(args, kwargs, json_data=json_data) - obj_, updated = self._data_layer.delete_relationship(json_data, - model_relationship_field, - related_id_field, - kwargs) + obj_, updated = self._data_layer.delete_relationship( + json_data, model_relationship_field, related_id_field, kwargs + ) status_code = 200 - result = {'meta': {'message': 'Relationship successfully updated'}} + result = {"meta": {"message": "Relationship successfully updated"}} if updated is False: - result = '' + result = "" status_code = 204 final_result = self.after_delete(result, status_code) @@ -521,16 +633,25 @@ def delete(self, *args, **kwargs): def _get_relationship_data(self): """Get useful data for relationship management""" - relationship_field = request.path.split('/')[-1].replace('-', '_') + relationship_field = request.path.split("/")[-1].replace("-", "_") if relationship_field not in get_relationships(self.schema): - raise RelationNotFound("{} has no attribute {}".format(self.schema.__name__, relationship_field)) + raise RelationNotFound( + "{} has no attribute {}".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 model_relationship_field = get_model_field(self.schema, relationship_field) - return relationship_field, model_relationship_field, related_type_, related_id_field + return ( + relationship_field, + model_relationship_field, + related_type_, + related_id_field, + ) def before_get(self, args, kwargs): """Hook to make custom work before get method""" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f5717ca --- /dev/null +++ b/requirements.txt @@ -0,0 +1,16 @@ +blinker==1.8.2 +click==8.1.7 +Flask==3.0.3 +flask_sqlalchemy==3.1.1 +iniconfig==2.0.0 +itsdangerous==2.2.0 +Jinja2==3.1.4 +MarkupSafe==2.1.5 +packaging==24.0 +pluggy==1.5.0 +pytest==8.2.1 +ruff==0.4.4 +uv==0.1.45 +Werkzeug==3.0.3 +marshmallow_jsonapi==0.24.0 +six==1.16.0 \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index 8bd3ed8..e95c568 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,6 @@ # -*- coding: utf-8 -*- import pytest - from flask import Flask @@ -11,6 +10,12 @@ def app(): return app +@pytest.fixture(scope="function") +def new_app(): + app = Flask(__name__) + return app + + @pytest.yield_fixture(scope="session") def client(app): return app.test_client() diff --git a/tests/test_sqlalchemy_data_layer.py b/tests/test_sqlalchemy_data_layer.py index 3662ca8..a84a5fd 100644 --- a/tests/test_sqlalchemy_data_layer.py +++ b/tests/test_sqlalchemy_data_layer.py @@ -1,28 +1,38 @@ # -*- coding: utf-8 -*- -from six.moves.urllib.parse import urlencode, parse_qs import pytest - -from sqlalchemy import create_engine, Column, Integer, DateTime, String, ForeignKey -from sqlalchemy.orm import sessionmaker, relationship -from sqlalchemy.ext.declarative import declarative_base -from flask import Blueprint, make_response, json -from marshmallow_jsonapi.flask import Schema, Relationship +from flask import Blueprint, json, make_response from marshmallow import Schema as MarshmallowSchema -from marshmallow_jsonapi import fields from marshmallow import ValidationError +from marshmallow_jsonapi import fields +from marshmallow_jsonapi.flask import Relationship, Schema +from six.moves.urllib.parse import parse_qs, urlencode +from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, create_engine +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import relationship, sessionmaker -from flask_rest_jsonapi import Api, ResourceList, ResourceDetail, ResourceRelationship, JsonApiException -from flask_rest_jsonapi.pagination import add_pagination_links -from flask_rest_jsonapi.exceptions import RelationNotFound, InvalidSort, InvalidFilters, InvalidInclude, BadRequest -from flask_rest_jsonapi.querystring import QueryStringManager as QSManager -from flask_rest_jsonapi.data_layers.alchemy import SqlalchemyDataLayer -from flask_rest_jsonapi.data_layers.base import BaseDataLayer -from flask_rest_jsonapi.data_layers.filtering.alchemy import Node -from flask_rest_jsonapi.decorators import check_headers, check_method_requirements, jsonapi_exception_formatter import flask_rest_jsonapi.decorators import flask_rest_jsonapi.resource import flask_rest_jsonapi.schema +from flask_rest_jsonapi import ( + Api, + JsonApiException, + ResourceDetail, + ResourceList, + ResourceRelationship, +) +from flask_rest_jsonapi.data_layers.alchemy import SqlalchemyDataLayer +from flask_rest_jsonapi.data_layers.base import BaseDataLayer +from flask_rest_jsonapi.data_layers.filtering.alchemy import Node +from flask_rest_jsonapi.exceptions import ( + BadRequest, + InvalidFilters, + InvalidInclude, + InvalidSort, + RelationNotFound, +) +from flask_rest_jsonapi.pagination import add_pagination_links +from flask_rest_jsonapi.querystring import QueryStringManager as QSManager @pytest.fixture(scope="module") @@ -33,9 +43,11 @@ def base(): @pytest.fixture(scope="module") def person_tag_model(base): class Person_Tag(base): - __tablename__ = 'person_tag' + __tablename__ = "person_tag" - id = Column(Integer, ForeignKey('person.person_id'), primary_key=True, index=True) + id = Column( + Integer, ForeignKey("person.person_id"), primary_key=True, index=True + ) key = Column(String, primary_key=True) value = Column(String, primary_key=True) @@ -45,9 +57,11 @@ class Person_Tag(base): @pytest.fixture(scope="module") def person_single_tag_model(base): class Person_Single_Tag(base): - __tablename__ = 'person_single_tag' + __tablename__ = "person_single_tag" - id = Column(Integer, ForeignKey('person.person_id'), primary_key=True, index=True) + id = Column( + Integer, ForeignKey("person.person_id"), primary_key=True, index=True + ) key = Column(String) value = Column(String) @@ -60,9 +74,10 @@ def string_json_attribute_person_model(base): This approach to faking JSON support for testing with sqlite is borrowed from: https://avacariu.me/articles/2016/compiling-json-as-text-for-sqlite-with-sqlalchemy """ - import sqlalchemy.types as types import json + import sqlalchemy.types as types + class StringyJSON(types.TypeDecorator): """Stores and retrieves JSON as TEXT.""" @@ -80,11 +95,10 @@ def process_result_value(self, value, dialect): # TypeEngine.with_variant says "use StringyJSON instead when # connecting to 'sqlite'" - MagicJSON = types.JSON().with_variant(StringyJSON, 'sqlite') + MagicJSON = types.JSON().with_variant(StringyJSON, "sqlite") class StringJsonAttributePerson(base): - - __tablename__ = 'string_json_attribute_person' + __tablename__ = "string_json_attribute_person" person_id = Column(Integer, primary_key=True) name = Column(String, nullable=False) @@ -99,15 +113,20 @@ class StringJsonAttributePerson(base): @pytest.fixture(scope="module") def person_model(base): class Person(base): - __tablename__ = 'person' + __tablename__ = "person" person_id = Column(Integer, primary_key=True) name = Column(String, nullable=False) birth_date = Column(DateTime) computers = relationship("Computer", backref="person") - tags = relationship("Person_Tag", cascade="save-update, merge, delete, delete-orphan") - single_tag = relationship("Person_Single_Tag", uselist=False, - cascade="save-update, merge, delete, delete-orphan") + tags = relationship( + "Person_Tag", cascade="save-update, merge, delete, delete-orphan" + ) + single_tag = relationship( + "Person_Single_Tag", + uselist=False, + cascade="save-update, merge, delete, delete-orphan", + ) yield Person @@ -115,17 +134,23 @@ class Person(base): @pytest.fixture(scope="module") def computer_model(base): class Computer(base): - __tablename__ = 'computer' + __tablename__ = "computer" id = Column(Integer, primary_key=True) serial = Column(String, nullable=False) - person_id = Column(Integer, ForeignKey('person.person_id')) + person_id = Column(Integer, ForeignKey("person.person_id")) yield Computer @pytest.fixture(scope="module") -def engine(person_tag_model, person_single_tag_model, person_model, computer_model, string_json_attribute_person_model): +def engine( + person_tag_model, + person_single_tag_model, + person_model, + computer_model, + string_json_attribute_person_model, +): engine = create_engine("sqlite:///:memory:") person_tag_model.metadata.create_all(engine) person_single_tag_model.metadata.create_all(engine) @@ -143,7 +168,7 @@ def session(engine): @pytest.fixture() def person(session, person_model): - person_ = person_model(name='test') + person_ = person_model(name="test") session_ = session session_.add(person_) session_.commit() @@ -154,7 +179,7 @@ def person(session, person_model): @pytest.fixture() def person_2(session, person_model): - person_ = person_model(name='test2') + person_ = person_model(name="test2") session_ = session session_.add(person_) session_.commit() @@ -165,7 +190,7 @@ def person_2(session, person_model): @pytest.fixture() def computer(session, computer_model): - computer_ = computer_model(serial='1') + computer_ = computer_model(serial="1") session_ = session session_.add(computer_) session_.commit() @@ -189,7 +214,7 @@ def wrapper_f(*args, **kwargs): def person_tag_schema(): class PersonTagSchema(MarshmallowSchema): class Meta: - type_ = 'person_tag' + type_ = "person_tag" id = fields.Str(load_only=True) key = fields.Str() @@ -202,7 +227,7 @@ class Meta: def person_single_tag_schema(): class PersonSingleTagSchema(MarshmallowSchema): class Meta: - type_ = 'person_single_tag' + type_ = "person_single_tag" id = fields.Str(load_only=True) key = fields.Str() @@ -216,7 +241,7 @@ def address_schema(): class AddressSchema(MarshmallowSchema): street = fields.String(required=True) city = fields.String(required=True) - state = fields.String(missing='NC') + state = fields.String(missing="NC") zip = fields.String(required=True) yield AddressSchema @@ -226,11 +251,11 @@ class AddressSchema(MarshmallowSchema): def string_json_attribute_person_schema(address_schema): class StringJsonAttributePersonSchema(Schema): class Meta: - type_ = 'string_json_attribute_person' - self_view = 'api.string_json_attribute_person_detail' - self_view_kwargs = {'person_id': ''} + type_ = "string_json_attribute_person" + self_view = "api.string_json_attribute_person_detail" + self_view_kwargs = {"person_id": ""} - id = fields.Integer(as_string=True, attribute='person_id') + id = fields.Integer(as_string=True, attribute="person_id") name = fields.Str(required=True) birth_date = fields.DateTime() address = fields.Nested(address_schema, many=False) @@ -242,18 +267,18 @@ class Meta: def person_schema(person_tag_schema, person_single_tag_schema): class PersonSchema(Schema): class Meta: - type_ = 'person' - self_view = 'api.person_detail' - self_view_kwargs = {'person_id': ''} + type_ = "person" + self_view = "api.person_detail" + self_view_kwargs = {"person_id": ""} - id = fields.Integer(as_string=True, attribute='person_id') + id = fields.Integer(as_string=True, attribute="person_id") name = fields.Str(required=True) birth_date = fields.DateTime() computers = Relationship( - related_view='api.computer_list', - related_view_kwargs={'person_id': ''}, - schema='ComputerSchema', - type_='computer', + related_view="api.computer_list", + related_view_kwargs={"person_id": ""}, + schema="ComputerSchema", + type_="computer", many=True, ) @@ -267,20 +292,22 @@ class Meta: def computer_schema(): class ComputerSchema(Schema): class Meta: - type_ = 'computer' - self_view = 'api.computer_detail' - self_view_kwargs = {'id': ''} + type_ = "computer" + self_view = "api.computer_detail" + self_view_kwargs = {"id": ""} id = fields.Integer(as_string=True) serial = fields.Str(required=True) - owner = Relationship(attribute='person', - default=None, - missing=None, - related_view='api.person_detail', - related_view_kwargs={'person_id': ''}, - schema='PersonSchema', - id_field='person_id', - type_='person') + owner = Relationship( + attribute="person", + default=None, + missing=None, + related_view="api.person_detail", + related_view_kwargs={"person_id": ""}, + schema="PersonSchema", + id_field="person_id", + type_="person", + ) yield ComputerSchema @@ -310,12 +337,16 @@ def before_delete_object_(self, obj, view_kwargs): @pytest.fixture(scope="module") -def person_list(session, person_model, dummy_decorator, person_schema, before_create_object): +def person_list( + session, person_model, dummy_decorator, person_schema, before_create_object +): class PersonList(ResourceList): schema = person_schema - data_layer = {'model': person_model, - 'session': session, - 'mzthods': {'before_create_object': before_create_object}} + data_layer = { + "model": person_model, + "session": session, + "mzthods": {"before_create_object": before_create_object}, + } get_decorators = [dummy_decorator] post_decorators = [dummy_decorator] get_schema_kwargs = dict() @@ -325,14 +356,25 @@ class PersonList(ResourceList): @pytest.fixture(scope="module") -def person_detail(session, person_model, dummy_decorator, person_schema, before_update_object, before_delete_object): +def person_detail( + session, + person_model, + dummy_decorator, + person_schema, + before_update_object, + before_delete_object, +): class PersonDetail(ResourceDetail): schema = person_schema - data_layer = {'model': person_model, - 'session': session, - 'url_field': 'person_id', - 'methods': {'before_update_object': before_update_object, - 'before_delete_object': before_delete_object}} + data_layer = { + "model": person_model, + "session": session, + "url_field": "person_id", + "methods": { + "before_update_object": before_update_object, + "before_delete_object": before_delete_object, + }, + } get_decorators = [dummy_decorator] patch_decorators = [dummy_decorator] delete_decorators = [dummy_decorator] @@ -347,9 +389,11 @@ class PersonDetail(ResourceDetail): def person_computers(session, person_model, dummy_decorator, person_schema): class PersonComputersRelationship(ResourceRelationship): schema = person_schema - data_layer = {'session': session, - 'model': person_model, - 'url_field': 'person_id'} + data_layer = { + "session": session, + "model": person_model, + "url_field": "person_id", + } get_decorators = [dummy_decorator] post_decorators = [dummy_decorator] patch_decorators = [dummy_decorator] @@ -362,11 +406,32 @@ class PersonComputersRelationship(ResourceRelationship): def person_list_raise_jsonapiexception(): class PersonList(ResourceList): def get(self): - raise JsonApiException('', '') + raise JsonApiException("", "") yield PersonList +@pytest.fixture(scope="module") +def person_computers_owned_relationship(person_schema, person_computers): + class PersonOwnedSchema(person_schema): + class Meta: + exclude = ("computers",) + + computers_owned = Relationship( + related_view="api.computer_list", + related_view_kwargs={"person_id": ""}, + schema="ComputerSchema", + type_="computer", + many=True, + attribute="computers", + ) + + class PersonComputersOwnedRelationship(person_computers): + schema = PersonOwnedSchema + + yield PersonComputersOwnedRelationship + + @pytest.fixture(scope="module") def person_list_raise_exception(): class PersonList(ResourceList): @@ -380,7 +445,7 @@ def get(self): def person_list_response(): class PersonList(ResourceList): def get(self): - return make_response('') + return make_response("") yield PersonList @@ -388,11 +453,10 @@ def get(self): @pytest.fixture(scope="module") def person_list_without_schema(session, person_model): class PersonList(ResourceList): - data_layer = {'model': person_model, - 'session': session} + data_layer = {"model": person_model, "session": session} def get(self): - return make_response('') + return make_response("") yield PersonList @@ -400,8 +464,12 @@ def get(self): @pytest.fixture(scope="module") def query(): def query_(self, view_kwargs): - if view_kwargs.get('person_id') is not None: - return self.session.query(computer_model).join(person_model).filter_by(person_id=view_kwargs['person_id']) + if view_kwargs.get("person_id") is not None: + return ( + self.session.query(computer_model) + .join(person_model) + .filter_by(person_id=view_kwargs["person_id"]) + ) return self.session.query(computer_model) yield query_ @@ -411,9 +479,11 @@ def query_(self, view_kwargs): def computer_list(session, computer_model, computer_schema, query): class ComputerList(ResourceList): schema = computer_schema - data_layer = {'model': computer_model, - 'session': session, - 'methods': {'query': query}} + data_layer = { + "model": computer_model, + "session": session, + "methods": {"query": query}, + } yield ComputerList @@ -422,9 +492,8 @@ class ComputerList(ResourceList): def computer_detail(session, computer_model, dummy_decorator, computer_schema): class ComputerDetail(ResourceDetail): schema = computer_schema - data_layer = {'model': computer_model, - 'session': session} - methods = ['GET', 'PATCH'] + data_layer = {"model": computer_model, "session": session} + methods = ["GET", "PATCH"] yield ComputerDetail @@ -433,36 +502,36 @@ class ComputerDetail(ResourceDetail): def computer_owner(session, computer_model, dummy_decorator, computer_schema): class ComputerOwnerRelationship(ResourceRelationship): schema = computer_schema - data_layer = {'session': session, - 'model': computer_model} + data_layer = {"session": session, "model": computer_model} yield ComputerOwnerRelationship @pytest.fixture(scope="module") -def string_json_attribute_person_detail(session, string_json_attribute_person_model, - string_json_attribute_person_schema): +def string_json_attribute_person_detail( + session, string_json_attribute_person_model, string_json_attribute_person_schema +): class StringJsonAttributePersonDetail(ResourceDetail): schema = string_json_attribute_person_schema - data_layer = {'session': session, - 'model': string_json_attribute_person_model} + data_layer = {"session": session, "model": string_json_attribute_person_model} yield StringJsonAttributePersonDetail @pytest.fixture(scope="module") -def string_json_attribute_person_list(session, string_json_attribute_person_model, string_json_attribute_person_schema): +def string_json_attribute_person_list( + session, string_json_attribute_person_model, string_json_attribute_person_schema +): class StringJsonAttributePersonList(ResourceList): schema = string_json_attribute_person_schema - data_layer = {'session': session, - 'model': string_json_attribute_person_model} + data_layer = {"session": session, "model": string_json_attribute_person_model} yield StringJsonAttributePersonList @pytest.fixture(scope="module") def api_blueprint(client): - bp = Blueprint('api', __name__) + bp = Blueprint("api", __name__) yield bp @@ -472,37 +541,95 @@ def api(api_blueprint): @pytest.fixture(scope="module") -def register_routes(client, api, app, api_blueprint, person_list, person_detail, person_computers, - person_list_raise_jsonapiexception, person_list_raise_exception, person_list_response, - person_list_without_schema, computer_list, computer_detail, computer_owner, - string_json_attribute_person_detail, string_json_attribute_person_list): - api.route(person_list, 'person_list', '/persons') - api.route(person_detail, 'person_detail', '/persons/') - api.route(person_computers, 'person_computers', '/persons//relationships/computers') - api.route(person_computers, 'person_computers_error', '/persons//relationships/computer') - api.route(person_list_raise_jsonapiexception, 'person_list_jsonapiexception', '/persons_jsonapiexception') - api.route(person_list_raise_exception, 'person_list_exception', '/persons_exception') - api.route(person_list_response, 'person_list_response', '/persons_response') - api.route(person_list_without_schema, 'person_list_without_schema', '/persons_without_schema') - api.route(computer_list, 'computer_list', '/computers', '/persons//computers') - api.route(computer_list, 'computer_detail', '/computers/') - api.route(computer_owner, 'computer_owner', '/computers//relationships/owner') - api.route(string_json_attribute_person_list, 'string_json_attribute_person_list', '/string_json_attribute_persons') - api.route(string_json_attribute_person_detail, 'string_json_attribute_person_detail', - '/string_json_attribute_persons/') +def register_routes( + client, + api, + app, + api_blueprint, + person_list, + person_detail, + person_computers, + person_list_raise_jsonapiexception, + person_list_raise_exception, + person_list_response, + person_list_without_schema, + computer_list, + computer_detail, + computer_owner, + string_json_attribute_person_detail, + string_json_attribute_person_list, + person_computers_owned_relationship, + request, +): + api.route(person_list, "person_list", "/persons") + api.route(person_detail, "person_detail", "/persons/") + api.route( + person_computers, + "person_computers", + "/persons//relationships/computers", + ) + api.route( + person_computers, + "person_computers_error", + "/persons//relationships/computer", + ) + api.route( + person_list_raise_jsonapiexception, + "person_list_jsonapiexception", + "/persons_jsonapiexception", + ) + api.route( + person_list_raise_exception, "person_list_exception", "/persons_exception" + ) + api.route(person_list_response, "person_list_response", "/persons_response") + api.route( + person_list_without_schema, + "person_list_without_schema", + "/persons_without_schema", + ) + api.route( + computer_list, + "computer_list", + "/computers", + "/persons//computers", + ) + api.route(computer_list, "computer_detail", "/computers/") + api.route( + computer_owner, "computer_owner", "/computers//relationships/owner" + ) + api.route( + string_json_attribute_person_list, + "string_json_attribute_person_list", + "/string_json_attribute_persons", + ) + api.route( + string_json_attribute_person_detail, + "string_json_attribute_person_detail", + "/string_json_attribute_persons/", + ) + api.route( + person_computers_owned_relationship, + "person_computers_owned", + "/persons//relationships/computers-owned", + ) + api.init_app(app) @pytest.fixture(scope="module") def get_object_mock(): class get_object(object): - foo = type('foo', (object,), { - 'property': type('prop', (object,), { - 'mapper': type('map', (object,), { - 'class_': 'test' - })() - })() - })() + foo = type( + "foo", + (object,), + { + "property": type( + "prop", + (object,), + {"mapper": type("map", (object,), {"class_": "test"})()}, + )() + }, + )() def __init__(self, kwargs): pass @@ -512,31 +639,29 @@ def __init__(self, kwargs): def test_add_pagination_links(app): with app.app_context(): - qs = {'page[number]': '2', 'page[size]': '10'} + qs = {"page[number]": "2", "page[size]": "10"} qsm = QSManager(qs, None) pagination_dict = dict() add_pagination_links(pagination_dict, 43, qsm, str()) - last_page_dict = parse_qs(pagination_dict['links']['last'][1:]) - assert len(last_page_dict['page[number]']) == 1 - assert last_page_dict['page[number]'][0] == '5' + last_page_dict = parse_qs(pagination_dict["links"]["last"][1:]) + assert len(last_page_dict["page[number]"]) == 1 + assert last_page_dict["page[number]"][0] == "5" def test_Node(person_model, person_schema, monkeypatch): from copy import deepcopy + filt = { - 'val': '0000', - 'field': True, - 'not': dict(), - 'name': 'name', - 'op': 'eq', - 'strip': lambda: 's' + "val": "0000", + "field": True, + "not": dict(), + "name": "name", + "op": "eq", + "strip": lambda: "s", } - filt['not'] = deepcopy(filt) - del filt['not']['not'] - n = Node(person_model, - filt, - None, - person_schema) + filt["not"] = deepcopy(filt) + del filt["not"]["not"] + n = Node(person_model, filt, None, person_schema) with pytest.raises(TypeError): # print(n.val is None and n.field is None) # # n.column @@ -546,7 +671,7 @@ def test_Node(person_model, person_schema, monkeypatch): n.column with pytest.raises(InvalidFilters): n.model = person_model - n.filter_['op'] = '' + n.filter_["op"] = "" n.operator with pytest.raises(InvalidFilters): n.related_model @@ -555,23 +680,23 @@ def test_Node(person_model, person_schema, monkeypatch): def test_check_method_requirements(monkeypatch): - self = type('self', (object,), dict()) - request = type('request', (object,), dict(method='GET')) - monkeypatch.setattr(flask_rest_jsonapi.decorators, 'request', request) + self = type("self", (object,), dict()) + request = type("request", (object,), dict(method="GET")) + monkeypatch.setattr(flask_rest_jsonapi.decorators, "request", request) with pytest.raises(Exception): flask_rest_jsonapi.decorators.check_method_requirements(lambda: 1)(self()) def test_json_api_exception(): - JsonApiException(None, None, title='test', status='test') + JsonApiException(None, None, title="test", status="test") def test_query_string_manager(person_schema): - query_string = {'page[slumber]': '3'} + query_string = {"page[slumber]": "3"} qsm = QSManager(query_string, person_schema) with pytest.raises(BadRequest): qsm.pagination - qsm.qs['sort'] = 'computers' + qsm.qs["sort"] = "computers" with pytest.raises(InvalidSort): qsm.sorting @@ -581,13 +706,14 @@ def schema_load_mock(*args, **kwargs): raise ValidationError(dict(errors=[dict(status=None, title=None)])) with app.app_context(): - query_string = {'page[slumber]': '3'} - app = type('app', (object,), dict(config=dict(DEBUG=True))) - headers = {'Content-Type': 'application/vnd.api+json'} - request = type('request', (object,), dict(method='POST', - headers=headers, - get_json=dict, - args=query_string)) + query_string = {"page[slumber]": "3"} + app = type("app", (object,), dict(config=dict(DEBUG=True))) + headers = {"Content-Type": "application/vnd.api+json"} + request = type( + "request", + (object,), + dict(method="POST", headers=headers, get_json=dict, args=query_string), + ) dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) rl = ResourceList() rd = ResourceDetail() @@ -595,12 +721,13 @@ def schema_load_mock(*args, **kwargs): rl.schema = person_schema rd._data_layer = dl rd.schema = person_schema - monkeypatch.setattr(flask_rest_jsonapi.resource, 'request', request) - monkeypatch.setattr(flask_rest_jsonapi.decorators, 'current_app', app) - monkeypatch.setattr(flask_rest_jsonapi.decorators, 'request', request) - monkeypatch.setattr(rl.schema, 'load', schema_load_mock) - r = super(flask_rest_jsonapi.resource.Resource, ResourceList) \ - .__new__(ResourceList) + monkeypatch.setattr(flask_rest_jsonapi.resource, "request", request) + monkeypatch.setattr(flask_rest_jsonapi.decorators, "current_app", app) + monkeypatch.setattr(flask_rest_jsonapi.decorators, "request", request) + monkeypatch.setattr(rl.schema, "load", schema_load_mock) + r = super(flask_rest_jsonapi.resource.Resource, ResourceList).__new__( + ResourceList + ) with pytest.raises(Exception): r.dispatch_request() rl.post() @@ -608,265 +735,273 @@ def schema_load_mock(*args, **kwargs): def test_compute_schema(person_schema): - query_string = {'page[number]': '3', 'fields[person]': list()} + query_string = {"page[number]": "3", "fields[person]": list()} qsm = QSManager(query_string, person_schema) with pytest.raises(InvalidInclude): - flask_rest_jsonapi.schema.compute_schema(person_schema, dict(), qsm, ['id']) - flask_rest_jsonapi.schema.compute_schema(person_schema, dict(only=list()), qsm, list()) + flask_rest_jsonapi.schema.compute_schema(person_schema, dict(), qsm, ["id"]) + flask_rest_jsonapi.schema.compute_schema( + person_schema, dict(only=list()), qsm, list() + ) def test_compute_schema_propagate_context(person_schema, computer_schema): query_string = {} qsm = QSManager(query_string, person_schema) - schema = flask_rest_jsonapi.schema.compute_schema(person_schema, dict(), qsm, ['computers']) - assert schema.declared_fields['computers'].__dict__['_Relationship__schema'].__dict__['context'] == dict() - schema = flask_rest_jsonapi.schema.compute_schema(person_schema, dict(context=dict(foo='bar')), qsm, ['computers']) - assert schema.declared_fields['computers'].__dict__['_Relationship__schema'].__dict__['context'] == dict(foo='bar') + schema = flask_rest_jsonapi.schema.compute_schema( + person_schema, dict(), qsm, ["computers"] + ) + assert ( + schema.declared_fields["computers"] + .__dict__["_Relationship__schema"] + .__dict__["context"] + == dict() + ) + schema = flask_rest_jsonapi.schema.compute_schema( + person_schema, dict(context=dict(foo="bar")), qsm, ["computers"] + ) + assert schema.declared_fields["computers"].__dict__[ + "_Relationship__schema" + ].__dict__["context"] == dict(foo="bar") # test good cases def test_get_list(client, register_routes, person, person_2): with client: - querystring = urlencode({ - 'page[number]': 1, - 'page[size]': 1, - 'fields[person]': 'name,birth_date', - 'sort': '-name', - 'include': 'computers.owner', - 'filter': json.dumps( - [ - { - 'and': [ - { - 'name': 'computers', - 'op': 'any', - 'val': { - 'name': 'serial', - 'op': 'eq', - 'val': '0000' - } - }, - { - 'or': [ - { - 'name': 'name', - 'op': 'like', - 'val': '%test%' + querystring = urlencode( + { + "page[number]": 1, + "page[size]": 1, + "fields[person]": "name,birth_date", + "sort": "-name", + "include": "computers.owner", + "filter": json.dumps( + [ + { + "and": [ + { + "name": "computers", + "op": "any", + "val": { + "name": "serial", + "op": "eq", + "val": "0000", }, - { - 'name': 'name', - 'op': 'like', - 'val': '%test2%' - } - ] - } - ] - } - ]) - }) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + }, + { + "or": [ + {"name": "name", "op": "like", "val": "%test%"}, + { + "name": "name", + "op": "like", + "val": "%test2%", + }, + ] + }, + ] + } + ] + ), + } + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] def test_get_list_with_simple_filter(client, register_routes, person, person_2): with client: - querystring = urlencode({'page[number]': 1, - 'page[size]': 1, - 'fields[person]': 'name,birth_date', - 'sort': '-name', - 'filter[name]': 'test' - }) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + querystring = urlencode( + { + "page[number]": 1, + "page[size]": 1, + "fields[person]": "name,birth_date", + "sort": "-name", + "filter[name]": "test", + } + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] def test_get_list_disable_pagination(client, register_routes): with client: - querystring = urlencode({'page[size]': 0}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + querystring = urlencode({"page[size]": 0}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] def test_head_list(client, register_routes): with client: - response = client.head('/persons', content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.head("/persons", content_type="application/vnd.api+json") + assert response.status_code == 200, response.json["errors"] def test_post_list(client, register_routes, computer): payload = { - 'data': { - 'type': 'person', - 'attributes': { - 'name': 'test' + "data": { + "type": "person", + "attributes": {"name": "test"}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.post('/persons', data=json.dumps(payload), content_type='application/vnd.api+json') - assert response.status_code == 201, response.json['errors'] + response = client.post( + "/persons", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 201, response.json["errors"] def test_post_list_nested_no_join(client, register_routes, computer): payload = { - 'data': { - 'type': 'string_json_attribute_person', - 'attributes': { - 'name': 'test_name', - 'address': { - 'street': 'test_street', - 'city': 'test_city', - 'state': 'NC', - 'zip': '00000' - } - } + "data": { + "type": "string_json_attribute_person", + "attributes": { + "name": "test_name", + "address": { + "street": "test_street", + "city": "test_city", + "state": "NC", + "zip": "00000", + }, + }, } } with client: - response = client.post('/string_json_attribute_persons', data=json.dumps(payload), - content_type='application/vnd.api+json') + response = client.post( + "/string_json_attribute_persons", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) print(response.get_data()) - assert response.status_code == 201, response.json['errors'] - assert json.loads(response.get_data())['data']['attributes']['address']['street'] == 'test_street' + assert response.status_code == 201, response.json["errors"] + assert ( + json.loads(response.get_data())["data"]["attributes"]["address"]["street"] + == "test_street" + ) def test_post_list_nested(client, register_routes, computer): payload = { - 'data': { - 'type': 'person', - 'attributes': { - 'name': 'test', - 'tags': [ - {'key': 'k1', 'value': 'v1'}, - {'key': 'k2', 'value': 'v2'} - ] + "data": { + "type": "person", + "attributes": { + "name": "test", + "tags": [{"key": "k1", "value": "v1"}, {"key": "k2", "value": "v2"}], + }, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.post('/persons', data=json.dumps(payload), content_type='application/vnd.api+json') - assert response.status_code == 201, response.json['errors'] - assert json.loads(response.get_data())['data']['attributes']['tags'][0]['key'] == 'k1' + response = client.post( + "/persons", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 201, response.json["errors"] + assert ( + json.loads(response.get_data())["data"]["attributes"]["tags"][0]["key"] + == "k1" + ) def test_post_list_single(client, register_routes, person): payload = { - 'data': { - 'type': 'computer', - 'attributes': { - 'serial': '1' + "data": { + "type": "computer", + "attributes": {"serial": "1"}, + "relationships": { + "owner": {"data": {"type": "person", "id": str(person.person_id)}} }, - 'relationships': { - 'owner': { - 'data': { - 'type': 'person', - 'id': str(person.person_id) - } - } - } } } with client: - response = client.post('/computers', data=json.dumps(payload), content_type='application/vnd.api+json') - assert response.status_code == 201, response.json['errors'] + response = client.post( + "/computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 201, response.json["errors"] def test_get_detail(client, register_routes, person): with client: - response = client.get('/persons/' + str(person.person_id), content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons/" + str(person.person_id), content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] def test_patch_detail(client, register_routes, computer, person): payload = { - 'data': { - 'id': str(person.person_id), - 'type': 'person', - 'attributes': { - 'name': 'test2' + "data": { + "id": str(person.person_id), + "type": "person", + "attributes": {"name": "test2"}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.patch('/persons/' + str(person.person_id), - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.patch( + "/persons/" + str(person.person_id), + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_patch_detail_nested(client, register_routes, computer, person): payload = { - 'data': { - 'id': str(person.person_id), - 'type': 'person', - 'attributes': { - 'name': 'test2', - 'tags': [ - {'key': 'new_key', 'value': 'new_value'} - ], - 'single_tag': {'key': 'new_single_key', 'value': 'new_single_value'} + "data": { + "id": str(person.person_id), + "type": "person", + "attributes": { + "name": "test2", + "tags": [{"key": "new_key", "value": "new_value"}], + "single_tag": {"key": "new_single_key", "value": "new_single_value"}, + }, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.patch('/persons/' + str(person.person_id), - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.patch( + "/persons/" + str(person.person_id), + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] response_dict = json.loads(response.get_data()) - assert response_dict['data']['attributes']['tags'][0]['key'] == 'new_key' - assert response_dict['data']['attributes']['single_tag']['key'] == 'new_single_key' + assert response_dict["data"]["attributes"]["tags"][0]["key"] == "new_key" + assert ( + response_dict["data"]["attributes"]["single_tag"]["key"] == "new_single_key" + ) def test_delete_detail(client, register_routes, person): with client: - response = client.delete('/persons/' + str(person.person_id), content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.delete( + "/persons/" + str(person.person_id), content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] def test_get_relationship(session, client, register_routes, computer, person): @@ -875,16 +1010,24 @@ def test_get_relationship(session, client, register_routes, computer, person): session_.commit() with client: - response = client.get('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_get_relationship_empty(client, register_routes, person): with client: - response = client.get('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_get_relationship_single(session, client, register_routes, computer, person): @@ -893,91 +1036,88 @@ def test_get_relationship_single(session, client, register_routes, computer, per session_.commit() with client: - response = client.get('/computers/' + str(computer.id) + '/relationships/owner', - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/computers/" + str(computer.id) + "/relationships/owner", + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_get_relationship_single_empty(session, client, register_routes, computer): with client: - response = client.get('/computers/' + str(computer.id) + '/relationships/owner', - content_type='application/vnd.api+json') + response = client.get( + "/computers/" + str(computer.id) + "/relationships/owner", + content_type="application/vnd.api+json", + ) response_json = json.loads(response.get_data()) - assert None is response_json['data'] - assert response.status_code == 200, response.json['errors'] + assert None is response_json["data"] + assert response.status_code == 200, response.json["errors"] def test_issue_49(session, client, register_routes, person, person_2): with client: for p in [person, person_2]: - response = client.get('/persons/' + str(p.person_id) + '/relationships/computers?include=computers', - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] - assert (json.loads(response.get_data()))['links']['related'] == '/persons/' + str( - p.person_id) + '/computers' + response = client.get( + "/persons/" + + str(p.person_id) + + "/relationships/computers?include=computers", + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] + assert (json.loads(response.get_data()))["links"][ + "related" + ] == "/persons/" + str(p.person_id) + "/computers" def test_post_relationship(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"type": "computer", "id": str(computer.id)}]} with client: - response = client.post('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.post( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_post_relationship_not_list(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'person', - 'id': str(person.person_id) - } - } + payload = {"data": {"type": "person", "id": str(person.person_id)}} with client: - response = client.post('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.post( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_patch_relationship(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"type": "computer", "id": str(computer.id)}]} with client: - response = client.patch('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.patch( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_patch_relationship_single(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'person', - 'id': str(person.person_id) - } - } + payload = {"data": {"type": "person", "id": str(person.person_id)}} with client: - response = client.patch('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.patch( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_delete_relationship(session, client, register_routes, computer, person): @@ -985,20 +1125,17 @@ def test_delete_relationship(session, client, register_routes, computer, person) person.computers = [computer] session_.commit() - payload = { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"type": "computer", "id": str(computer.id)}]} with client: - response = client.delete('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.delete( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_delete_relationship_single(session, client, register_routes, computer, person): @@ -1006,93 +1143,116 @@ def test_delete_relationship_single(session, client, register_routes, computer, computer.person = person session_.commit() - payload = { - 'data': { - 'type': 'person', - 'id': str(person.person_id) - } - } + payload = {"data": {"type": "person", "id": str(person.person_id)}} with client: - response = client.delete('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.delete( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"] def test_get_list_response(client, register_routes): with client: - response = client.get('/persons_response', content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons_response", content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] class TestResourceArgs: - def test_resource_args(self, app): + def test_resource_args(self, new_app): class TestResource(ResourceDetail): """ This fake resource always renders a constructor parameter """ + def __init__(self, *args, **kwargs): super(TestResource, self).__init__() self.constant = args[0] def get(self): return self.constant - api = Api(app=app) - api.route(TestResource, 'resource_args', '/resource_args', resource_args=['hello!']) + + api = Api(app=new_app) + api.route( + TestResource, "resource_args", "/resource_args", resource_args=["hello!"] + ) api.init_app() - with app.test_client() as client: - rv = client.get('/resource_args') - assert rv.json == 'hello!' + with new_app.test_client() as client: + rv = client.get("/resource_args") + assert rv.json == "hello!" - def test_resource_kwargs(self, app): + def test_resource_kwargs(self, new_app): class TestResource(ResourceDetail): """ This fake resource always renders a constructor parameter """ + def __init__(self, *args, **kwargs): super(TestResource, self).__init__() - self.constant = kwargs.get('constant') + self.constant = kwargs.get("constant") def get(self): return self.constant - api = Api(app=app) - api.route(TestResource, 'resource_kwargs', '/resource_kwargs', resource_kwargs={ - 'constant': 'hello!' - }) + + api = Api(app=new_app) + api.route( + TestResource, + "resource_kwargs", + "/resource_kwargs", + resource_kwargs={"constant": "hello!"}, + ) api.init_app() - with app.test_client() as client: - rv = client.get('/resource_kwargs') - assert rv.json == 'hello!' + with new_app.test_client() as client: + rv = client.get("/resource_kwargs") + assert rv.json == "hello!" # test various Accept headers def test_single_accept_header(client, register_routes): with client: - response = client.get('/persons', content_type='application/vnd.api+json', - headers={'Accept': 'application/vnd.api+json'}) - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons", + content_type="application/vnd.api+json", + headers={"Accept": "application/vnd.api+json"}, + ) + assert response.status_code == 200, response.json["errors"] def test_multiple_accept_header(client, register_routes): with client: - response = client.get('/persons', content_type='application/vnd.api+json', - headers={'Accept': '*/*, application/vnd.api+json, application/vnd.api+json;q=0.9'}) - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons", + content_type="application/vnd.api+json", + headers={ + "Accept": "*/*, application/vnd.api+json, application/vnd.api+json;q=0.9" + }, + ) + assert response.status_code == 200, response.json["errors"] def test_wrong_accept_header(client, register_routes): with client: - response = client.get('/persons', content_type='application/vnd.api+json', - headers={'Accept': 'application/vnd.api+json;q=0.7, application/vnd.api+json;q=0.9'}) - assert response.status_code == 406, response.json['errors'] + response = client.get( + "/persons", + content_type="application/vnd.api+json", + headers={ + "Accept": "application/vnd.api+json;q=0.7, application/vnd.api+json;q=0.9" + }, + ) + assert response.status_code == 406, response.json["errors"] # test Content-Type error def test_wrong_content_type(client, register_routes): with client: - response = client.post('/persons', headers={'Content-Type': 'application/vnd.api+json;q=0.8'}) - assert response.status_code == 415, response.json['errors'] + response = client.post( + "/persons", headers={"Content-Type": "application/vnd.api+json;q=0.8"} + ) + assert response.status_code == 415, response.json["errors"] @pytest.fixture(scope="module") @@ -1105,14 +1265,16 @@ class WrongDataLayer(object): def test_wrong_data_layer_inheritence(wrong_data_layer): with pytest.raises(Exception): + class PersonDetail(ResourceDetail): - data_layer = {'class': wrong_data_layer} + data_layer = {"class": wrong_data_layer} PersonDetail() def test_wrong_data_layer_kwargs_type(): with pytest.raises(Exception): + class PersonDetail(ResourceDetail): data_layer = list() @@ -1121,134 +1283,175 @@ class PersonDetail(ResourceDetail): def test_get_list_jsonapiexception(client, register_routes): with client: - response = client.get('/persons_jsonapiexception', content_type='application/vnd.api+json') - assert response.status_code == 500, response.json['errors'] + response = client.get( + "/persons_jsonapiexception", content_type="application/vnd.api+json" + ) + assert response.status_code == 500, response.json["errors"] def test_get_list_exception(client, register_routes): with client: - response = client.get('/persons_exception', content_type='application/vnd.api+json') - assert response.status_code == 500, response.json['errors'] + response = client.get( + "/persons_exception", content_type="application/vnd.api+json" + ) + assert response.status_code == 500, response.json["errors"] def test_get_list_without_schema(client, register_routes): with client: - response = client.post('/persons_without_schema', content_type='application/vnd.api+json') - assert response.status_code == 500, response.json['errors'] + response = client.post( + "/persons_without_schema", content_type="application/vnd.api+json" + ) + assert response.status_code == 500, response.json["errors"] def test_get_list_bad_request(client, register_routes): with client: - querystring = urlencode({'page[number': 3}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"page[number": 3}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_invalid_fields(client, register_routes): with client: - querystring = urlencode({'fields[person]': 'error'}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"fields[person]": "error"}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_invalid_include(client, register_routes): with client: - querystring = urlencode({'include': 'error'}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"include": "error"}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_invalid_filters_parsing(client, register_routes): with client: - querystring = urlencode({'filter': 'error'}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"filter": "error"}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_invalid_page(client, register_routes): with client: - querystring = urlencode({'page[number]': 'error'}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"page[number]": "error"}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_invalid_sort(client, register_routes): with client: - querystring = urlencode({'sort': 'error'}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"sort": "error"}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_detail_object_not_found(client, register_routes): with client: - response = client.get('/persons/3', content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get("/persons/3", content_type="application/vnd.api+json") + assert response.status_code == 200, response.json["errors"] def test_post_relationship_related_object_not_found(client, register_routes, person): - payload = { - 'data': [ - { - 'type': 'computer', - 'id': '2' - } - ] - } + payload = {"data": [{"type": "computer", "id": "2"}]} with client: - response = client.post('/persons/' + str(person.person_id) + '/relationships/computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 404, response.json['errors'] + response = client.post( + "/persons/" + str(person.person_id) + "/relationships/computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 404, response.json["errors"] def test_get_relationship_relationship_field_not_found(client, register_routes, person): with client: - response = client.get('/persons/' + str(person.person_id) + '/relationships/computer', - content_type='application/vnd.api+json') - assert response.status_code == 500, response.json['errors'] + response = client.get( + "/persons/" + str(person.person_id) + "/relationships/computer", + content_type="application/vnd.api+json", + ) + assert response.status_code == 500, response.json["errors"] def test_get_list_invalid_filters_val(client, register_routes): with client: - querystring = urlencode({'filter': json.dumps([{'name': 'computers', 'op': 'any'}])}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode( + {"filter": json.dumps([{"name": "computers", "op": "any"}])} + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_name(client, register_routes): with client: - querystring = urlencode({'filter': json.dumps([{'name': 'computers__serial', 'op': 'any', 'val': '1'}])}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + querystring = urlencode( + { + "filter": json.dumps( + [{"name": "computers__serial", "op": "any", "val": "1"}] + ) + } + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 200, response.json["errors"] def test_get_list_no_name(client, register_routes): with client: - querystring = urlencode({'filter': json.dumps([{'op': 'any', 'val': '1'}])}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode({"filter": json.dumps([{"op": "any", "val": "1"}])}) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_no_op(client, register_routes): with client: - querystring = urlencode({'filter': json.dumps([{'name': 'computers__serial', 'val': '1'}])}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode( + {"filter": json.dumps([{"name": "computers__serial", "val": "1"}])} + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_attr_error(client, register_routes): with client: - querystring = urlencode({'filter': json.dumps([{'name': 'error', 'op': 'eq', 'val': '1'}])}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode( + {"filter": json.dumps([{"name": "error", "op": "eq", "val": "1"}])} + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_get_list_field_error(client, register_routes): with client: - querystring = urlencode({'filter': json.dumps([{'name': 'name', 'op': 'eq', 'field': 'error'}])}) - response = client.get('/persons' + '?' + querystring, content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + querystring = urlencode( + {"filter": json.dumps([{"name": "name", "op": "eq", "field": "error"}])} + ) + response = client.get( + "/persons" + "?" + querystring, content_type="application/vnd.api+json" + ) + assert response.status_code == 400, response.json["errors"] def test_sqlalchemy_data_layer_without_session(person_model, person_list): @@ -1263,27 +1466,37 @@ def test_sqlalchemy_data_layer_without_model(session, person_list): def test_sqlalchemy_data_layer_create_object_error(session, person_model, person_list): with pytest.raises(JsonApiException): - dl = SqlalchemyDataLayer(dict(session=session, model=person_model, resource=person_list)) + dl = SqlalchemyDataLayer( + dict(session=session, model=person_model, resource=person_list) + ) dl.create_object(dict(), dict()) def test_sqlalchemy_data_layer_get_object_error(session, person_model): with pytest.raises(Exception): - dl = SqlalchemyDataLayer(dict(session=session, model=person_model, id_field='error')) + dl = SqlalchemyDataLayer( + dict(session=session, model=person_model, id_field="error") + ) dl.get_object(dict()) -def test_sqlalchemy_data_layer_update_object_error(session, person_model, person_list, monkeypatch): +def test_sqlalchemy_data_layer_update_object_error( + session, person_model, person_list, monkeypatch +): def commit_mock(): raise JsonApiException() with pytest.raises(JsonApiException): - dl = SqlalchemyDataLayer(dict(session=session, model=person_model, resource=person_list)) - monkeypatch.setattr(dl.session, 'commit', commit_mock) + dl = SqlalchemyDataLayer( + dict(session=session, model=person_model, resource=person_list) + ) + monkeypatch.setattr(dl.session, "commit", commit_mock) dl.update_object(dict(), dict(), dict()) -def test_sqlalchemy_data_layer_delete_object_error(session, person_model, person_list, monkeypatch): +def test_sqlalchemy_data_layer_delete_object_error( + session, person_model, person_list, monkeypatch +): def commit_mock(): raise JsonApiException() @@ -1291,532 +1504,516 @@ def delete_mock(obj): pass with pytest.raises(JsonApiException): - dl = SqlalchemyDataLayer(dict(session=session, model=person_model, resource=person_list)) - monkeypatch.setattr(dl.session, 'commit', commit_mock) - monkeypatch.setattr(dl.session, 'delete', delete_mock) + dl = SqlalchemyDataLayer( + dict(session=session, model=person_model, resource=person_list) + ) + monkeypatch.setattr(dl.session, "commit", commit_mock) + monkeypatch.setattr(dl.session, "delete", delete_mock) dl.delete_object(dict(), dict()) -def test_sqlalchemy_data_layer_create_relationship_field_not_found(session, person_model): +def test_sqlalchemy_data_layer_create_relationship_field_not_found( + session, person_model +): with pytest.raises(Exception): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - dl.create_relationship(dict(), 'error', '', dict(id=1)) + dl.create_relationship(dict(), "error", "", dict(id=1)) -def test_sqlalchemy_data_layer_create_relationship_error(session, person_model, get_object_mock, monkeypatch): +def test_sqlalchemy_data_layer_create_relationship_error( + session, person_model, get_object_mock, monkeypatch +): def commit_mock(): raise JsonApiException() with pytest.raises(JsonApiException): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - monkeypatch.setattr(dl.session, 'commit', commit_mock) - monkeypatch.setattr(dl, 'get_object', get_object_mock) - dl.create_relationship(dict(data=None), 'foo', '', dict(id=1)) + monkeypatch.setattr(dl.session, "commit", commit_mock) + monkeypatch.setattr(dl, "get_object", get_object_mock) + dl.create_relationship(dict(data=None), "foo", "", dict(id=1)) def test_sqlalchemy_data_layer_get_relationship_field_not_found(session, person_model): with pytest.raises(RelationNotFound): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - dl.get_relationship('error', '', '', dict(id=1)) + dl.get_relationship("error", "", "", dict(id=1)) -def test_sqlalchemy_data_layer_update_relationship_field_not_found(session, person_model): +def test_sqlalchemy_data_layer_update_relationship_field_not_found( + session, person_model +): with pytest.raises(Exception): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - dl.update_relationship(dict(), 'error', '', dict(id=1)) + dl.update_relationship(dict(), "error", "", dict(id=1)) -def test_sqlalchemy_data_layer_update_relationship_error(session, person_model, get_object_mock, monkeypatch): +def test_sqlalchemy_data_layer_update_relationship_error( + session, person_model, get_object_mock, monkeypatch +): def commit_mock(): raise JsonApiException() with pytest.raises(JsonApiException): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - monkeypatch.setattr(dl.session, 'commit', commit_mock) - monkeypatch.setattr(dl, 'get_object', get_object_mock) - dl.update_relationship(dict(data=None), 'foo', '', dict(id=1)) + monkeypatch.setattr(dl.session, "commit", commit_mock) + monkeypatch.setattr(dl, "get_object", get_object_mock) + dl.update_relationship(dict(data=None), "foo", "", dict(id=1)) -def test_sqlalchemy_data_layer_delete_relationship_field_not_found(session, person_model): +def test_sqlalchemy_data_layer_delete_relationship_field_not_found( + session, person_model +): with pytest.raises(Exception): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - dl.delete_relationship(dict(), 'error', '', dict(id=1)) + dl.delete_relationship(dict(), "error", "", dict(id=1)) -def test_sqlalchemy_data_layer_delete_relationship_error(session, person_model, get_object_mock, monkeypatch): +def test_sqlalchemy_data_layer_delete_relationship_error( + session, person_model, get_object_mock, monkeypatch +): def commit_mock(): raise JsonApiException() with pytest.raises(JsonApiException): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - monkeypatch.setattr(dl.session, 'commit', commit_mock) - monkeypatch.setattr(dl, 'get_object', get_object_mock) - dl.delete_relationship(dict(data=None), 'foo', '', dict(id=1)) + monkeypatch.setattr(dl.session, "commit", commit_mock) + monkeypatch.setattr(dl, "get_object", get_object_mock) + dl.delete_relationship(dict(data=None), "foo", "", dict(id=1)) def test_sqlalchemy_data_layer_sort_query_error(session, person_model, monkeypatch): with pytest.raises(InvalidSort): dl = SqlalchemyDataLayer(dict(session=session, model=person_model)) - dl.sort_query(None, [dict(field='test')]) + dl.sort_query(None, [dict(field="test")]) def test_post_list_incorrect_type(client, register_routes, computer): payload = { - 'data': { - 'type': 'error', - 'attributes': { - 'name': 'test' + "data": { + "type": "error", + "attributes": {"name": "test"}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.post('/persons', data=json.dumps(payload), content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.post( + "/persons", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_post_list_validation_error(client, register_routes, computer): payload = { - 'data': { - 'type': 'person', - 'attributes': {}, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } + "data": { + "type": "person", + "attributes": {}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} + }, } } with client: - response = client.post('/persons', data=json.dumps(payload), content_type='application/vnd.api+json') - assert response.status_code == 422, response.json['errors'] + response = client.post( + "/persons", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 422, response.json["errors"] def test_patch_detail_incorrect_type(client, register_routes, computer, person): payload = { - 'data': { - 'id': str(person.person_id), - 'type': 'error', - 'attributes': { - 'name': 'test2' + "data": { + "id": str(person.person_id), + "type": "error", + "attributes": {"name": "test2"}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.patch('/persons/' + str(person.person_id), - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.patch( + "/persons/" + str(person.person_id), + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_patch_detail_validation_error(client, register_routes, computer, person): payload = { - 'data': { - 'id': str(person.person_id), - 'type': 'person', - 'attributes': { - 'name': {'test2': 'error'} + "data": { + "id": str(person.person_id), + "type": "person", + "attributes": {"name": {"test2": "error"}}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.patch('/persons/' + str(person.person_id), - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 422, response.json['errors'] + response = client.patch( + "/persons/" + str(person.person_id), + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 422, response.json["errors"] def test_patch_detail_missing_id(client, register_routes, computer, person): payload = { - 'data': { - 'type': 'person', - 'attributes': { - 'name': 'test2' + "data": { + "type": "person", + "attributes": {"name": "test2"}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.patch('/persons/' + str(person.person_id), - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.patch( + "/persons/" + str(person.person_id), + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_patch_detail_wrong_id(client, register_routes, computer, person): payload = { - 'data': { - 'id': 'error', - 'type': 'person', - 'attributes': { - 'name': 'test2' + "data": { + "id": "error", + "type": "person", + "attributes": {"name": "test2"}, + "relationships": { + "computers": {"data": [{"type": "computer", "id": str(computer.id)}]} }, - 'relationships': { - 'computers': { - 'data': [ - { - 'type': 'computer', - 'id': str(computer.id) - } - ] - } - } } } with client: - response = client.patch('/persons/' + str(person.person_id), - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 422, response.json['errors'] + response = client.patch( + "/persons/" + str(person.person_id), + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 422, response.json["errors"] def test_post_relationship_no_data(client, register_routes, computer, person): with client: - response = client.post('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(dict()), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.post( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(dict()), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_post_relationship_not_list_missing_type(client, register_routes, computer, person): - payload = { - 'data': { - 'id': str(person.person_id) - } - } +def test_post_relationship_not_list_missing_type( + client, register_routes, computer, person +): + payload = {"data": {"id": str(person.person_id)}} with client: - response = client.post('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.post( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_post_relationship_not_list_missing_id(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'person' - } - } +def test_post_relationship_not_list_missing_id( + client, register_routes, computer, person +): + payload = {"data": {"type": "person"}} with client: - response = client.post('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.post( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_post_relationship_not_list_wrong_type(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'error', - 'id': str(person.person_id) - } - } +def test_post_relationship_not_list_wrong_type( + client, register_routes, computer, person +): + payload = {"data": {"type": "error", "id": str(person.person_id)}} with client: - response = client.post('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.post( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_post_relationship_missing_type(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"id": str(computer.id)}]} with client: - response = client.post('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.post( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_post_relationship_missing_id(client, register_routes, computer, person): payload = { - 'data': [ + "data": [ { - 'type': 'computer', + "type": "computer", } ] } with client: - response = client.post('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.post( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_post_relationship_wrong_type(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'type': 'error', - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"type": "error", "id": str(computer.id)}]} with client: - response = client.post('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.post( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_patch_relationship_no_data(client, register_routes, computer, person): with client: - response = client.patch('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(dict()), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.patch( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(dict()), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_patch_relationship_not_list_missing_type(client, register_routes, computer, person): - payload = { - 'data': { - 'id': str(person.person_id) - } - } +def test_patch_relationship_not_list_missing_type( + client, register_routes, computer, person +): + payload = {"data": {"id": str(person.person_id)}} with client: - response = client.patch('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.patch( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_patch_relationship_not_list_missing_id(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'person' - } - } +def test_patch_relationship_not_list_missing_id( + client, register_routes, computer, person +): + payload = {"data": {"type": "person"}} with client: - response = client.patch('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.patch( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_patch_relationship_not_list_wrong_type(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'error', - 'id': str(person.person_id) - } - } +def test_patch_relationship_not_list_wrong_type( + client, register_routes, computer, person +): + payload = {"data": {"type": "error", "id": str(person.person_id)}} with client: - response = client.patch('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.patch( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_patch_relationship_missing_type(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"id": str(computer.id)}]} with client: - response = client.patch('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.patch( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_patch_relationship_missing_id(client, register_routes, computer, person): payload = { - 'data': [ + "data": [ { - 'type': 'computer', + "type": "computer", } ] } with client: - response = client.patch('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.patch( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_patch_relationship_wrong_type(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'type': 'error', - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"type": "error", "id": str(computer.id)}]} with client: - response = client.patch('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.patch( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_delete_relationship_no_data(client, register_routes, computer, person): with client: - response = client.delete('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(dict()), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.delete( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(dict()), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_delete_relationship_not_list_missing_type(client, register_routes, computer, person): - payload = { - 'data': { - 'id': str(person.person_id) - } - } +def test_delete_relationship_not_list_missing_type( + client, register_routes, computer, person +): + payload = {"data": {"id": str(person.person_id)}} with client: - response = client.delete('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.delete( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_delete_relationship_not_list_missing_id(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'person' - } - } +def test_delete_relationship_not_list_missing_id( + client, register_routes, computer, person +): + payload = {"data": {"type": "person"}} with client: - response = client.delete('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.delete( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] -def test_delete_relationship_not_list_wrong_type(client, register_routes, computer, person): - payload = { - 'data': { - 'type': 'error', - 'id': str(person.person_id) - } - } +def test_delete_relationship_not_list_wrong_type( + client, register_routes, computer, person +): + payload = {"data": {"type": "error", "id": str(person.person_id)}} with client: - response = client.delete('/computers/' + str(computer.id) + '/relationships/owner', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.delete( + "/computers/" + str(computer.id) + "/relationships/owner", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_delete_relationship_missing_type(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"id": str(computer.id)}]} with client: - response = client.delete('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.delete( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_delete_relationship_missing_id(client, register_routes, computer, person): payload = { - 'data': [ + "data": [ { - 'type': 'computer', + "type": "computer", } ] } with client: - response = client.delete('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 400, response.json['errors'] + response = client.delete( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 400, response.json["errors"] def test_delete_relationship_wrong_type(client, register_routes, computer, person): - payload = { - 'data': [ - { - 'type': 'error', - 'id': str(computer.id) - } - ] - } + payload = {"data": [{"type": "error", "id": str(computer.id)}]} with client: - response = client.delete('/persons/' + str(person.person_id) + '/relationships/computers?include=computers', - data=json.dumps(payload), - content_type='application/vnd.api+json') - assert response.status_code == 409, response.json['errors'] + response = client.delete( + "/persons/" + + str(person.person_id) + + "/relationships/computers?include=computers", + data=json.dumps(payload), + content_type="application/vnd.api+json", + ) + assert response.status_code == 409, response.json["errors"] def test_base_data_layer(): @@ -1884,46 +2081,54 @@ def test_qs_manager(): QSManager([], None) -def test_api(app, person_list): - api = Api(app) - api.route(person_list, 'person_list', '/persons', '/person_list') +def test_api(new_app, person_list): + api = Api(new_app) + api.route(person_list, "person_list", "/persons", "/person_list") api.init_app() -def test_api_resources(app, person_list): +def test_api_resources(new_app, person_list): api = Api() - api.route(person_list, 'person_list2', '/persons', '/person_list') - api.init_app(app) - - -def test_relationship_containing_hyphens(api, app, client, person_schema, person_computers, register_routes, - computer_schema, person): + api.route(person_list, "person_list2", "/persons", "/person_list") + api.init_app(new_app) + + +def test_relationship_containing_hyphens( + api, + app, + client, + person_schema, + person_computers, + register_routes, + computer_schema, + person, +): """ This is a bit of a hack. Basically, since we can no longer have two attributes that read from the same key in Marshmallow 3, we have to create a new Schema and Resource here that name their relationship "computers_owned" - in order to test hyphenation + in order to test hyphenation. + + See `PersonOwnedSchema` and `person_computers_owned_relationship` """ class PersonOwnedSchema(person_schema): class Meta: - exclude = ('computers',) + exclude = ("computers",) computers_owned = Relationship( - related_view='api.computer_list', - related_view_kwargs={'person_id': ''}, - schema='ComputerSchema', - type_='computer', + related_view="api.computer_list", + related_view_kwargs={"person_id": ""}, + schema="ComputerSchema", + type_="computer", many=True, - attribute='computers' + attribute="computers", ) class PersonComputersOwnedRelationship(person_computers): schema = PersonOwnedSchema - api.route(PersonComputersOwnedRelationship, 'person_computers_owned', - '/persons//relationships/computers-owned') - api.init_app(app) - - response = client.get('/persons/{}/relationships/computers-owned'.format(person.person_id), - content_type='application/vnd.api+json') - assert response.status_code == 200, response.json['errors'] + response = client.get( + "/persons/{}/relationships/computers-owned".format(person.person_id), + content_type="application/vnd.api+json", + ) + assert response.status_code == 200, response.json["errors"]