Created
October 3, 2013 17:04
-
-
Save rugginoso/6813244 to your computer and use it in GitHub Desktop.
Initial implementation of Resource class for RESTfull services
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from flask import Flask, abort | |
from resource import Resource | |
app = Flask(__name__) | |
class Item(Resource): | |
def list(self, user_id, car_id): | |
pass | |
def detail(self, user_id, car_id, item_id): | |
pass | |
def create(self, user_id, car_id): | |
pass | |
def update(self, user_id, car_id, item_id): | |
pass | |
def remove(self, user_id, car_id, item_id): | |
pass | |
class Car(Resource): | |
nested_resources = [Item] | |
def list(self, user_id): | |
pass | |
def detail(self, user_id, car_id): | |
pass | |
class User(Resource): | |
nested_resources = [Car] | |
def list(self): | |
pass | |
def detail(self, user_id): | |
pass | |
def create(self): | |
pass | |
def update(self, user_id): | |
pass | |
def remove(self, user_id): | |
pass | |
for rule, kwargs in User.get_url_rules(): | |
app.add_url_rule(rule, **kwargs) | |
if __name__ == '__main__': | |
print('Rules:') | |
for rule in app.url_map.iter_rules(): | |
print(repr(rule)) | |
app.run(debug=True) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from flask import Flask, abort | |
from flask.views import MethodView, MethodViewType | |
class ResourceType(MethodViewType): | |
def __new__(cls, name, bases, dict): | |
dict['resource_name'] = dict.get('resource_name') or name.lower() | |
dict['resource_path'] = dict.get( | |
'resource_path') or '/%ss/' % dict['resource_name'] | |
dict['resource_parameter_type'] = dict.get( | |
'resource_parameter_type') or 'int' | |
dict['resource_parameter_name'] = dict.get( | |
'resource_parameter_name') or '%s_id' % dict['resource_name'] | |
return type.__new__(cls, name, bases, dict) | |
class Resource(MethodView): | |
__metaclass__ = ResourceType | |
COLLECTION_ACTIONS = { | |
'list': 'GET', | |
'create': 'POST', | |
} | |
ITEM_ACTIONS = { | |
'detail': 'GET', | |
'update': 'PUT', | |
'remove': 'DELETE', | |
} | |
nested_resources = [] | |
def get(self, *args, **kwargs): | |
if self.get_resource_parameter_name() in kwargs: | |
return self._try_action_or_403('detail', *args, **kwargs) | |
else: | |
return self._try_action_or_403('list', *args, **kwargs) | |
def post(self, *args, **kwargs): | |
return self._try_action_or_403('create', *args, **kwargs) | |
def put(self, *args, **kwargs): | |
return self._try_action_or_403('update', *args, **kwargs) | |
def delete(self, *args, **kwargs): | |
return self._try_action_or_403('remove', *args, **kwargs) | |
def _try_action_or_403(self, action, *args, **kwargs): | |
try: | |
action = getattr(self, action) | |
return action(*args, **kwargs) | |
except AttributeError: | |
abort(403) | |
@classmethod | |
def get_resource_parameter(cls): | |
return ( | |
'<%s:%s>' % ( | |
cls.resource_parameter_type, | |
cls.resource_parameter_name) | |
) | |
@classmethod | |
def get_url_rules(cls): | |
resource_parameter = cls.get_resource_parameter() | |
view_func = cls.as_view(cls.resource_name + '_api') | |
attrs = set(dir(cls)) | |
collection_methods = [ | |
method for action, | |
method in cls.COLLECTION_ACTIONS.items( | |
) if action in attrs] | |
item_methods = [ | |
method for action, | |
method in cls.ITEM_ACTIONS.items( | |
) if action in attrs] | |
rules = [] | |
if collection_methods: | |
rules.append( | |
(cls.resource_path, | |
{'view_func': view_func, | |
'methods': collection_methods})) | |
if item_methods: | |
rules.append( | |
(''.join((cls.resource_path, resource_parameter)), {'view_func': view_func, 'methods': item_methods})) | |
for nested_resource in cls.nested_resources: | |
for rule, kwargs in nested_resource.get_url_rules(): | |
rules.append( | |
(''.join((cls.resource_path, resource_parameter, rule)), kwargs)) | |
return rules |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment