3771624463

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #68 from AdCombo/fix__init_subclass__for_multi_project

Run Details

60 of 60 new or added lines in 1 file covered. (100.0%)

1351 of 1620 relevant lines covered (83.4%)

0.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

45.97

/flask_combo_jsonapi/api.py

"""This module contains the main class of the Api to initialize the Api, plug default decorators for each resources
methods, speficy which blueprint to use, define the Api routes and plug additional oauth manager and permission manager
"""

import inspect
from functools import wraps

from flask import request, abort

from flask_combo_jsonapi.decorators import jsonapi_exception_formatter
from flask_combo_jsonapi.exceptions import PluginMethodNotImplementedError
from flask_combo_jsonapi.resource import ResourceList, ResourceRelationship


class Api(object):
    """The main class of the Api"""

    def __init__(self, app=None, blueprint=None, decorators=None, plugins=None, qs_manager_class=None):
        """Initialize an instance of the Api

        :param app: the flask application
        :param blueprint: a flask blueprint
        :param tuple decorators: a tuple of decorators plugged to each resource methods
        :param plugins: list of plugins
        :param qs_manager_class: custom query string manager used in whole API
        """
        self.app = app
        self._app = app
        self.blueprint = blueprint
        self.resources = []
        self.resource_registry = []
        self.decorators = decorators or tuple()
        self.plugins = plugins if plugins is not None else []
        self.qs_manager_class = qs_manager_class

        if app is not None:
            self.init_app(app, blueprint)

    def init_app(self, app=None, blueprint=None, additional_blueprints=None):
        """Update flask application with our api

        :param Application app: a flask application
        :param blueprint:
        :param additional_blueprints:
        """
        if app is not None:
            self.app = app

        if blueprint is not None:
            self.blueprint = blueprint

        for resource in self.resources:
            self.route(resource['resource'],
                       resource['view'],
                       *resource['urls'],
                       url_rule_options=resource['url_rule_options'])

        if self.blueprint is not None:
            self.app.register_blueprint(self.blueprint)

        if additional_blueprints is not None:
            for blueprint in additional_blueprints:
                self.app.register_blueprint(blueprint)

        self.app.config.setdefault('PAGE_SIZE', 30)

        for i_plugin in self.plugins:
            try:
                i_plugin.after_init_plugin(app=None, blueprint=None, additional_blueprints=None)
            except PluginMethodNotImplementedError:
                pass

    def route(self, resource, view, *urls, **kwargs):
        """Create an api view.

        :param Resource resource: a resource class inherited from flask_combo_jsonapi.resource.Resource
        :param str view: the view name
        :param str urls: the urls of the view
        :param kwargs: additional options of the route
        """
        for i_plugin in self.plugins:
            try:
                i_plugin.before_route(resource=resource, view=view, urls=urls, self_json_api=self, **kwargs)
            except PluginMethodNotImplementedError:
                pass
        setattr(resource, 'plugins', self.plugins)

        if self.qs_manager_class:
            setattr(resource, 'qs_manager_class', self.qs_manager_class)

        resource.view = view
        url_rule_options = kwargs.get('url_rule_options') or dict()

        if hasattr(resource, 'decorators'):
            resource.decorators += self.decorators
        else:
            resource.decorators = self.decorators

        view_func = resource.as_view(view)

        if 'blueprint' in kwargs:
            resource.view = '.'.join([kwargs['blueprint'].name, resource.view])
            for url in urls:
                kwargs['blueprint'].add_url_rule(url, view_func=view_func, **url_rule_options)
        elif self.blueprint is not None:
            resource.view = '.'.join([self.blueprint.name, resource.view])
            for url in urls:
                self.blueprint.add_url_rule(url, view_func=view_func, **url_rule_options)
        elif self.app is not None:
            for url in urls:
                self.app.add_url_rule(url, view_func=view_func, **url_rule_options)
        else:
            self.resources.append({'resource': resource,
                                   'view': view,
                                   'urls': urls,
                                   'url_rule_options': url_rule_options})

        self.resource_registry.append(resource)

        for i_plugin in self.plugins:
            try:
                i_plugin.after_route(resource=resource, view=view, urls=urls, self_json_api=self, **kwargs)
            except PluginMethodNotImplementedError:
                pass

    def oauth_manager(self, oauth_manager):
        """Use the oauth manager to enable oauth for API

        :param oauth_manager: the oauth manager
        """
        @self.app.before_request
        @jsonapi_exception_formatter
        def before_request():
            endpoint = request.endpoint
            resource = None
            if endpoint:
                resource = getattr(self.app.view_functions[endpoint], 'view_class', None)

            if resource and not getattr(resource, 'disable_oauth', None):
                scopes = request.args.get('scopes')

                if getattr(resource, 'schema'):
                    scopes = [self.build_scope(resource, request.method)]
                elif scopes:
                    scopes = scopes.split(',')

                    if scopes:
                        scopes = scopes.split(',')

                valid, req = oauth_manager.verify_request(scopes)

                for func in oauth_manager._after_request_funcs:
                    valid, req = func(valid, req)

                if not valid:
                    if oauth_manager._invalid_response:
                        return oauth_manager._invalid_response(req)
                    return abort(401)

                request.oauth = req

    @staticmethod
    def build_scope(resource, method):
        """Compute the name of the scope for oauth

        :param Resource resource: the resource manager
        :param str method: an http method
        :return str: the name of the scope
        """
        if ResourceList in inspect.getmro(resource) and method == 'GET':
            prefix = 'list'
        else:
            method_to_prefix = {'GET': 'get',
                                'POST': 'create',
                                'PATCH': 'update',
                                'DELETE': 'delete'}
            prefix = method_to_prefix[method]

            if ResourceRelationship in inspect.getmro(resource):
                prefix = '_'.join([prefix, 'relationship'])

        return '_'.join([prefix, resource.schema.opts.type_])

    def permission_manager(self, permission_manager, with_decorators=True):
        """Use permission manager to enable permission for API

        :param callable permission_manager: the permission manager
        """
        self.check_permissions = permission_manager

        if with_decorators:
            for resource in self.resource_registry:
                if getattr(resource, 'disable_permission', None) is not True:
                    for method in getattr(resource, 'methods', ('GET', 'POST', 'PATCH', 'DELETE')):
                        setattr(resource,
                                method.lower(),
                                self.has_permission()(getattr(resource, method.lower())))

    def has_permission(self, *args, **kwargs):
        """Decorator used to check permissions before to call resource manager method"""
        def wrapper(view):
            if getattr(view, '_has_permissions_decorator', False) is True:
                return view

            @wraps(view)
            @jsonapi_exception_formatter
            def decorated(*view_args, **view_kwargs):
                self.check_permissions(view, view_args, view_kwargs, *args, **kwargs)
                return view(*view_args, **view_kwargs)
            decorated._has_permissions_decorator = True
            return decorated
        return wrapper

    @staticmethod
    def check_permissions(view, view_args, view_kwargs, *args, **kwargs):
        """The function use to check permissions

        :param callable view: the view
        :param list view_args: view args
        :param dict view_kwargs: view kwargs
        :param list args: decorator args
        :param dict kwargs: decorator kwargs
        """
        raise NotImplementedError

1	"""This module contains the main class of the Api to initialize the Api, plug default decorators for each resources
2	methods, speficy which blueprint to use, define the Api routes and plug additional oauth manager and permission manager
3	"""
4
5	import inspect	1✔
6	from functools import wraps	1✔
7
8	from flask import request, abort	1✔
9
10	from flask_combo_jsonapi.decorators import jsonapi_exception_formatter	1✔
11	from flask_combo_jsonapi.exceptions import PluginMethodNotImplementedError	1✔
12	from flask_combo_jsonapi.resource import ResourceList, ResourceRelationship	1✔
13
14
15	class Api(object):	1✔
16	"""The main class of the Api"""
17
18	def __init__(self, app=None, blueprint=None, decorators=None, plugins=None, qs_manager_class=None):	1✔
19	"""Initialize an instance of the Api
20
21	:param app: the flask application
22	:param blueprint: a flask blueprint
23	:param tuple decorators: a tuple of decorators plugged to each resource methods
24	:param plugins: list of plugins
25	:param qs_manager_class: custom query string manager used in whole API
26	"""
27	self.app = app	1✔
28	self._app = app	1✔
29	self.blueprint = blueprint	1✔
30	self.resources = []	1✔
31	self.resource_registry = []	1✔
32	self.decorators = decorators or tuple()	1✔
33	self.plugins = plugins if plugins is not None else []	1✔
34	self.qs_manager_class = qs_manager_class	1✔
35
36	if app is not None:	1✔
37	self.init_app(app, blueprint)	1✔
38
39	def init_app(self, app=None, blueprint=None, additional_blueprints=None):	1✔
40	"""Update flask application with our api
41
42	:param Application app: a flask application
43	:param blueprint:
44	:param additional_blueprints:
45	"""
46	if app is not None:	1✔
47	self.app = app	1✔
48
49	if blueprint is not None:	1✔
50	self.blueprint = blueprint	×
51
52	for resource in self.resources:	1✔
53	self.route(resource['resource'],	1✔
54	resource['view'],
55	*resource['urls'],
56	url_rule_options=resource['url_rule_options'])
57
58	if self.blueprint is not None:	1✔
59	self.app.register_blueprint(self.blueprint)	1✔
60
61	if additional_blueprints is not None:	1✔
62	for blueprint in additional_blueprints:	×
63	self.app.register_blueprint(blueprint)	×
64
65	self.app.config.setdefault('PAGE_SIZE', 30)	1✔
66
67	for i_plugin in self.plugins:	1✔
68	try:	×
69	i_plugin.after_init_plugin(app=None, blueprint=None, additional_blueprints=None)	×
70	except PluginMethodNotImplementedError:	×
71	pass	×
72
73	def route(self, resource, view, urls, *kwargs):	1✔
74	"""Create an api view.
75
76	:param Resource resource: a resource class inherited from flask_combo_jsonapi.resource.Resource
77	:param str view: the view name
78	:param str urls: the urls of the view
79	:param kwargs: additional options of the route
80	"""
81	for i_plugin in self.plugins:	1✔
82	try:	×
83	i_plugin.before_route(resource=resource, view=view, urls=urls, self_json_api=self, **kwargs)	×
84	except PluginMethodNotImplementedError:	×
85	pass	×
86	setattr(resource, 'plugins', self.plugins)	1✔
87
88	if self.qs_manager_class:	1✔
89	setattr(resource, 'qs_manager_class', self.qs_manager_class)	1✔
90
91	resource.view = view	1✔
92	url_rule_options = kwargs.get('url_rule_options') or dict()	1✔
93
94	if hasattr(resource, 'decorators'):	1✔
95	resource.decorators += self.decorators	1✔
96	else:
97	resource.decorators = self.decorators	×
98
99	view_func = resource.as_view(view)	1✔
100
101	if 'blueprint' in kwargs:	1✔
102	resource.view = '.'.join([kwargs['blueprint'].name, resource.view])	×
103	for url in urls:	×
104	kwargs['blueprint'].add_url_rule(url, view_func=view_func, **url_rule_options)	×
105	elif self.blueprint is not None:	1✔
106	resource.view = '.'.join([self.blueprint.name, resource.view])	1✔
107	for url in urls:	1✔
108	self.blueprint.add_url_rule(url, view_func=view_func, **url_rule_options)	1✔
109	elif self.app is not None:	1✔
110	for url in urls:	1✔
111	self.app.add_url_rule(url, view_func=view_func, **url_rule_options)	1✔
112	else:
113	self.resources.append({'resource': resource,	1✔
114	'view': view,
115	'urls': urls,
116	'url_rule_options': url_rule_options})
117
118	self.resource_registry.append(resource)	1✔
119
120	for i_plugin in self.plugins:	1✔
121	try:	×
122	i_plugin.after_route(resource=resource, view=view, urls=urls, self_json_api=self, **kwargs)	×
123	except PluginMethodNotImplementedError:	×
124	pass	×
125
126	def oauth_manager(self, oauth_manager):	1✔
127	"""Use the oauth manager to enable oauth for API
128
129	:param oauth_manager: the oauth manager
130	"""
131	@self.app.before_request	×
132	@jsonapi_exception_formatter	×
133	def before_request():	×
134	endpoint = request.endpoint	×
135	resource = None	×
136	if endpoint:	×
137	resource = getattr(self.app.view_functions[endpoint], 'view_class', None)	×
138
139	if resource and not getattr(resource, 'disable_oauth', None):	×
140	scopes = request.args.get('scopes')	×
141
142	if getattr(resource, 'schema'):	×
143	scopes = [self.build_scope(resource, request.method)]	×
144	elif scopes:	×
145	scopes = scopes.split(',')	×
146
147	if scopes:	×
148	scopes = scopes.split(',')	×
149
150	valid, req = oauth_manager.verify_request(scopes)	×
151
152	for func in oauth_manager._after_request_funcs:	×
153	valid, req = func(valid, req)	×
154
155	if not valid:	×
156	if oauth_manager._invalid_response:	×
157	return oauth_manager._invalid_response(req)	×
158	return abort(401)	×
159
160	request.oauth = req	×
161
162	@staticmethod	1✔
163	def build_scope(resource, method):	1✔
164	"""Compute the name of the scope for oauth
165
166	:param Resource resource: the resource manager
167	:param str method: an http method
168	:return str: the name of the scope
169	"""
170	if ResourceList in inspect.getmro(resource) and method == 'GET':	×
171	prefix = 'list'	×
172	else:
173	method_to_prefix = {'GET': 'get',	×
174	'POST': 'create',
175	'PATCH': 'update',
176	'DELETE': 'delete'}
177	prefix = method_to_prefix[method]	×
178
179	if ResourceRelationship in inspect.getmro(resource):	×
180	prefix = '_'.join([prefix, 'relationship'])	×
181
182	return '_'.join([prefix, resource.schema.opts.type_])	×
183
184	def permission_manager(self, permission_manager, with_decorators=True):	1✔
185	"""Use permission manager to enable permission for API
186
187	:param callable permission_manager: the permission manager
188	"""
189	self.check_permissions = permission_manager	×
190
191	if with_decorators:	×
192	for resource in self.resource_registry:	×
193	if getattr(resource, 'disable_permission', None) is not True:	×
194	for method in getattr(resource, 'methods', ('GET', 'POST', 'PATCH', 'DELETE')):	×
195	setattr(resource,	×
196	method.lower(),
197	self.has_permission()(getattr(resource, method.lower())))
198
199	def has_permission(self, args, *kwargs):	1✔
200	"""Decorator used to check permissions before to call resource manager method"""
201	def wrapper(view):	×
202	if getattr(view, '_has_permissions_decorator', False) is True:	×
203	return view	×
204
205	@wraps(view)	×
206	@jsonapi_exception_formatter	×
207	def decorated(view_args, *view_kwargs):	×
208	self.check_permissions(view, view_args, view_kwargs, args, *kwargs)	×
209	return view(view_args, *view_kwargs)	×
210	decorated._has_permissions_decorator = True	×
211	return decorated	×
212	return wrapper	×
213
214	@staticmethod	1✔
215	def check_permissions(view, view_args, view_kwargs, args, *kwargs):	1✔
216	"""The function use to check permissions
217
218	:param callable view: the view
219	:param list view_args: view args
220	:param dict view_kwargs: view kwargs
221	:param list args: decorator args
222	:param dict kwargs: decorator kwargs
223	"""
224	raise NotImplementedError	×

AdCombo / flask-combo-jsonapi / 3771624463

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous