Lately, I've had to develop REST APIs for several projects, sometimes in Java and sometimes in Python. Web services for a typical CRUD API usually have pretty much the same elements. Sooner or later that makes you think about how to reuse your code for different REST resources. I'm going to explain the approach that I've used for my latest Pyramid project. Let's take blog posts as a simple example:
- GET: /posts
- fetches a list of blog posts.
- GET: /posts/{id}
- reads a single post with the given ID.
- POST: /posts
- inserts a new post into the posts collection.
- PUT: /posts/{id}
- updates the post with the given ID.
- DELETE: /posts/{id}
- deletes the given post.
Let's say we add two routes:
config.add_route('posts_collection', '/posts')
config.add_route('posts', '/posts/{id}')
Then, a fairly basic implementation of the web services might look like this:
from pyramid.httpexceptions import HTTPOk
from pyramid.view import view_config, view_defaults
from sampleproject.models import DBSession, Post, PostSchema
@view_defaults(renderer='json', route_name='posts')
class PostView(object):
def __init__(self, request):
self.request = request
# Only load a single blog post when we have a post id:
if request.matched_route.name == 'posts':
post_id = int(request.matchdict['id'])
self.post = DBSession.query(Post).get(post_id)
@view_config(route_name='posts_collection', request_method='GET')
def list_posts(self):
return DBSession.query(Post).all()
@view_config(route_name='posts_collection', request_method='POST')
def create_post(self):
data = PostSchema().deserialize(self.request.json_body)
post = Post(**data)
DBSession.add(post)
# Flush to get the post.id from the database
DBSession.flush()
return post
@view_config(request_method='GET')
def read_post(self):
return self.post
@view_config(request_method='PUT')
def update_post(self):
data = PostSchema().deserialize(self.request.json_body)
# update the post data
for key, value in data.items():
setattr(self.post, key, value)
return self.post
@view_config(request_method='DELETE')
def delete_post(self):
DBSession.delete(self.post)
return HTTPOk()
The code above uses SQLAlchemy as ORM and Colander schema to validate incoming data. Changes to the threaded DBSession are automatically committed at the end of each request (in case you're wondering about the lack of DBSession.commit() statements.)
The Challenge
After a while you might need similar API code for comments, attachments, users or something else. You could simply repeat the code above and replace Post with Attachment, the 'posts' route with an 'attachments' route and so on. But that kind of code repetition is bad. So, in a good DRY manner, let's extract the common code to a base class to make it re-usable:
# all your imports
class BaseView(object):
item_cls = None # Override this in child views
schema_cls = None # Override this in child views
def __init__(self, request):
self.request = request
# Don't load a single item for the collection route
if request.matched_route.name == self.item_route:
item_id = int(request.matchdict['id'])
item = DBSession.query(self.item_cls).get(item_id)
self.item = item
def create_item(self):
data = self.schema_cls().deserialize(self.request.json_body)
item = self.item_cls(**data)
DBSession.add(item)
DBSession.flush()
return item
def list_items(self):
return DBSession.query(self.item_cls).all()
def read_item(self):
return self.item
def update_item(self):
data = self.schema_cls().deserialize(self.request.json_body)
for key, value in data.items():
setattr(self.item, key, value)
return self.item
def delete_item(self):
DBSession.delete(self.item)
return HTTPOk()
The idea is that we can now simply write this in our child classes:
class PostView(BaseView):
item_cls = Post
schema_cls = PostSchema
and the CRUD services are inherited from BaseView. But guess what, it won't work like this. We have to connect our view methods to our routes somehow, otherwise nothing happens at all. My first idea was to decorate the methods in BaseView with view_config:
class BaseView(object):
...
@view_config(request_method='GET')
def read_item(self):
return self.item
@view_config(request_method='PUT')
def update_item(self):
data = self.schema_cls().deserialize(self.request.json_body)
for key, value in data.items():
setattr(self.item, key, value)
return self.item
...
and then simply decorate the actual PostView with a view_defaults:
@view_defaults(route_name='posts', renderer='json')
class PostView(BaseView):
item_cls = Post
schema_cls = PostSchema
hoping that Pyramid would merge the route in the view_defaults with the request methods in the parent view_config. But that does not work either. It seems that Pyramid doesn't recognize view_config when it's only applied in parent classes.
The Solution
So, how can we register our views without repeating view_config for every single method in every child view? The answer is: We build ourselves a custom decorator called register_views that does all the view_config work for us:
class register_views(object):
def __init__(self, route=None, collection_route=None):
self.route = route
self.collection_route = collection_route
def __call__(self, cls):
cls.item_route = self.route
if self.route:
cls = view_config(_depth=1, renderer='json', attr='read_item',
request_method='GET', route_name=self.route)(cls)
cls = view_config(_depth=1, renderer='json', attr='update_item',
request_method='PUT', route_name=self.route)(cls)
cls = view_config(_depth=1, renderer='json', attr='delete_item',
request_method='DELETE', route_name=self.route)(cls)
if self.collection_route:
cls = view_config(_depth=1, renderer='json', attr='list_items',
request_method='GET', route_name=self.collection_route)(cls)
cls = view_config(_depth=1, renderer='json', attr='create_item',
request_method='POST', route_name=self.collection_route)(cls)
return cls
With this, we can create child views with a small snippet:
@register_views(route='posts', collection_route='posts_collection')
class PostView(BaseView):
item_cls = Post
schema_cls = PostSchema
The Explanation
A short explanation of what register_views does:
- The view_config decorator is applied imperatively, which means
@view_config(some_parameter="hello")
is replaced with
cls = view_config(some_parameter="hello")(cls)
- Instead of decorating every single method, we can apply view_config to the class itself with a parameter attr that specifies which method we want to register. Thus,
@view_config(request_method='GET')
def read_item(self):
...
becomes:
cls = view_config(request_method='GET', attr='read_item')(cls)
- We need to know two different routes for every child class. These are passed as arguments to the decorator and will be available in the __init__ constructor of the decorator:
@register_views(route='posts', collection_route='posts_collection')
- The approach above applies the view_config decorator within another decorator/function. In Pyramid that is tricky because it uses Venusian to do some decorator magic (which is rather advanced and which I haven't grasped fully yet). In order to tell Pyramid that our decorators are nested one level deep, we need to add the _depth=1 parameter.
In the end, that gives us this approach for the core of our register_views:
cls = view_config(_depth=1, renderer='json', attr='read_item',
request_method='GET', route_name=self.route)(cls)
Further reading: