from enum import Enum from typing import Callable, Iterable, Iterator, Tuple, Type, Union import inflection from anytree import PreOrderIter from boltons.typeutils import classproperty, issubclass from ereuse_devicehub.ereuse_utils.naming import Naming from flask import Blueprint, current_app, g, request, url_for from flask.json import jsonify from flask.views import MethodView from marshmallow import Schema as MarshmallowSchema from marshmallow import SchemaOpts as MarshmallowSchemaOpts from marshmallow import ValidationError, post_dump, pre_load, validates_schema from werkzeug.exceptions import MethodNotAllowed from werkzeug.routing import UnicodeConverter from ereuse_devicehub.teal import db, query class SchemaOpts(MarshmallowSchemaOpts): """ Subclass of Marshmallow's SchemaOpts that provides options for Teal's schemas. """ def __init__(self, meta, ordered=False): super().__init__(meta, ordered) self.PREFIX = meta.PREFIX class Schema(MarshmallowSchema): """ The definition of the fields of a resource. """ OPTIONS_CLASS = SchemaOpts class Meta: PREFIX = None """Optional. A prefix for the type; ex. devices:Computer.""" # noinspection PyMethodParameters @classproperty def t(cls: Type['Schema']) -> str: """The type for this schema, auto-computed from its name.""" name, *_ = cls.__name__.split('Schema') return Naming.new_type(name, cls.Meta.PREFIX) # noinspection PyMethodParameters @classproperty def resource(cls: Type['Schema']) -> str: """The resource name of this schema.""" return Naming.resource(cls.t) @validates_schema(pass_original=True) def check_unknown_fields(self, _, original_data: dict): """ Raises a validationError when user sends extra fields. From `Marshmallow docs`_. """ unknown_fields = set(original_data) - set( f.data_key or n for n, f in self.fields.items() ) if unknown_fields: raise ValidationError('Unknown field', unknown_fields) @validates_schema(pass_original=True) def check_dump_only(self, _, orig_data: dict): """ Raises a ValidationError if the user is submitting 'read-only' fields. """ # Note that validates_schema does not execute when dumping dump_only_fields = ( name for name, field in self.fields.items() if field.dump_only ) non_writable = set(orig_data).intersection(dump_only_fields) if non_writable: raise ValidationError('Non-writable field', non_writable) @pre_load @post_dump def remove_none_values(self, data: dict) -> dict: """ Skip from dumping and loading values that are None. A value that is None will be the same as a value that has not been set. `From here `_. """ # Will I always want this? # maybe this could be a setting in the future? return {key: value for key, value in data.items() if value is not None} def dump( self, model: Union['db.Model', Iterable['db.Model']], many=None, update_fields=True, nested=None, polymorphic_on='t', ): """ Like marshmallow's dump but with nested resource support and it only works for Models. This can load model relationships up to ``nested`` level. For example, if ``nested`` is ``1`` and we pass in a model of ``User`` that has a relationship with a table of ``Post``, it will load ``User`` and ``User.posts`` with all posts objects populated, but it won't load relationships inside the ``Post`` object. If, at the same time the ``Post`` has an ``author`` relationship with ``author_id`` being the FK, ``user.posts[n].author`` will be the value of ``author_id``. Define nested fields with the :class:`ereuse_devicehub.teal.marshmallow.NestedOn` This method requires an active application context as it needs to store some stuff in ``g``. :param nested: How many layers of nested relationships to load? By default only loads 1 nested relationship. """ from ereuse_devicehub.teal.marshmallow import NestedOn if nested is not None: setattr(g, NestedOn.NESTED_LEVEL, 0) setattr(g, NestedOn.NESTED_LEVEL_MAX, nested) if many: # todo this breaks with normal dicts. Maybe this should go # in NestedOn in the same way it happens when loading if isinstance(model, dict): return super().dump(model, update_fields=update_fields) else: return [ self._polymorphic_dump(o, update_fields, polymorphic_on) for o in model ] else: if isinstance(model, dict): return super().dump(model, update_fields=update_fields) else: return self._polymorphic_dump(model, update_fields, polymorphic_on) def _polymorphic_dump(self, obj: 'db.Model', update_fields, polymorphic_on='t'): schema = current_app.resources[getattr(obj, polymorphic_on)].schema if schema.t != self.t: return super(schema.__class__, schema).dump(obj, False, update_fields) else: return super().dump(obj, False, update_fields) def jsonify( self, model: Union['db.Model', Iterable['db.Model']], nested=1, many=False, update_fields: bool = True, polymorphic_on='t', **kw, ) -> str: """ Like flask's jsonify but with model / marshmallow schema support. :param nested: How many layers of nested relationships to load? By default only loads 1 nested relationship. """ return jsonify(self.dump(model, many, update_fields, nested, polymorphic_on)) class View(MethodView): """ A REST interface for resources. """ QUERY_PARSER = query.NestedQueryFlaskParser() class FindArgs(MarshmallowSchema): """ Allowed arguments for the ``find`` method (GET collection) endpoint """ def __init__(self, definition: 'Resource', **kw) -> None: self.resource_def = definition """The ResourceDefinition tied to this view.""" self.schema = None # type: Schema """The schema tied to this view.""" self.find_args = self.FindArgs() super().__init__() def dispatch_request(self, *args, **kwargs): # This is unique for each view call self.schema = g.schema """ The default schema in this resource. Added as an attr for commodity; you can always use g.schema. """ return super().dispatch_request(*args, **kwargs) def get(self, id): """Get a collection of resources or a specific one. --- parameters: - name: id in: path description: The identifier of the resource. type: string required: false responses: 200: description: Return the collection or the specific one. """ if id: response = self.one(id) else: args = self.QUERY_PARSER.parse( self.find_args, request, locations=('querystring',) ) response = self.find(args) return response def one(self, id): """GET one specific resource (ex. /cars/1).""" raise MethodNotAllowed() def find(self, args: dict): """GET a list of resources (ex. /cars).""" raise MethodNotAllowed() def post(self): raise MethodNotAllowed() def delete(self, id): raise MethodNotAllowed() def put(self, id): raise MethodNotAllowed() def patch(self, id): raise MethodNotAllowed() class Converters(Enum): """An enumeration of available URL converters.""" string = 'string' int = 'int' float = 'float' path = 'path' any = 'any' uuid = 'uuid' lower = 'lower' class LowerStrConverter(UnicodeConverter): """Like StringConverter but lowering the string.""" def to_python(self, value): return super().to_python(value).lower() class Resource(Blueprint): """ Main resource class. Defines the schema, views, authentication, database and collection of a resource. A ``ResourceDefinition`` is a Flask :class:`flask.blueprints.Blueprint` that provides everything needed to set a REST endpoint. """ VIEW = None # type: Type[View] """ Resource view linked to this definition or None. If none, this resource does not generate any view. """ SCHEMA = Schema # type: Type[Schema] """The Schema that validates a submitting resource at the entry point.""" AUTH = False """ If true, authentication is required for all the endpoints of this resource defined in ``VIEW``. """ ID_NAME = 'id' """ The variable name for GET *one* operations that is used as an id. """ ID_CONVERTER = Converters.string """ The converter for the id. Note that converters do **cast** the value, so the converter ``uuid`` will return an ``UUID`` object. """ __type__ = None # type: str """ The type of resource. If none, it is used the type of the Schema (``Schema.type``) """ def __init__( self, app, import_name=__name__, static_folder=None, static_url_path=None, template_folder=None, url_prefix=None, subdomain=None, url_defaults=None, root_path=None, cli_commands: Iterable[Tuple[Callable, str or None]] = tuple(), ): assert not self.VIEW or issubclass( self.VIEW, View ), 'VIEW should be a subclass of View' assert not self.SCHEMA or issubclass( self.SCHEMA, Schema ), 'SCHEMA should be a subclass of Schema or None.' # todo test for cases where self.SCHEMA is None url_prefix = ( url_prefix if url_prefix is not None else '/{}'.format(self.resource) ) super().__init__( self.type, import_name, static_folder, static_url_path, template_folder, url_prefix, subdomain, url_defaults, root_path, ) # todo __name__ in import_name forces subclasses to override the constructor # otherwise import_name equals to teal.resource not project1.myresource # and it is not very elegant... self.app = app self.schema = self.SCHEMA() if self.SCHEMA else None # Views if self.VIEW: view = self.VIEW.as_view('main', definition=self, auth=app.auth) if self.AUTH: view = app.auth.requires_auth(view) self.add_url_rule( '/', defaults={'id': None}, view_func=view, methods={'GET'} ) self.add_url_rule('/', view_func=view, methods={'POST'}) self.add_url_rule( '/<{}:{}>'.format(self.ID_CONVERTER.value, self.ID_NAME), view_func=view, methods={'GET', 'PUT', 'DELETE', 'PATCH'}, ) self.cli_commands = cli_commands self.before_request(self.load_resource) @classproperty def type(cls): t = cls.__type__ or cls.SCHEMA.t assert t, 'Resource needs a type: either from SCHEMA or manually from __type__.' return t @classproperty def t(cls): return cls.type @classproperty def resource(cls): return Naming.resource(cls.type) @classproperty def cli_name(cls): """The name used to generate the CLI Click group for this resource.""" return inflection.singularize(cls.resource) def load_resource(self): """ Loads a schema and resource_def into the current request so it can be used easily by functions outside view. """ g.schema = self.schema g.resource_def = self def init_db(self, db: 'db.SQLAlchemy', exclude_schema=None): """ Put here code to execute when initializing the database for this resource. We guarantee this to be executed in an app_context. No need to commit. """ pass @property def subresources_types(self) -> Iterator[str]: """Gets the types of the subresources.""" return (node.name for node in PreOrderIter(self.app.tree[self.t])) TYPE = Union[ Resource, Schema, 'db.Model', str, Type[Resource], Type[Schema], Type['db.Model'] ] def url_for_resource(resource: TYPE, item_id=None, method='GET') -> str: """ As Flask's ``url_for``, this generates an URL but specifically for a View endpoint of the given resource. :param method: The method whose view URL should be generated. :param resource: :param item_id: If given, append the ID of the resource in the URL, ex. GET /devices/1 :return: An URL. """ type = getattr(resource, 't', resource) values = {} if item_id: values[current_app.resources[type].ID_NAME] = item_id return url_for('{}.main'.format(type), _method=method, **values)