Cornice provides helpers to build and document REST-ish Web Services with Pyramid; and SQLAlchemy is the best Python ORM. I wanted to use Cornice and SQLAlchemy to make a simple RESTful webapp and couldn’t find any info on how to put them together.
Here’s how (full source in this github repo):
Make a simple Cornice application
Following the tutorial pretty closely.
$ mkvirtualenv blogpostcorniceapp
$ pip install cornice
$ pcreate -t blogpostcorniceapp
$ cd blogpostcorniceapp
$ python setup.py develop
$ pserve blogpostcorniceapp.ini
visit localhost:6543 -> {"Hello": "World"}
Excellent!
Now we can define a service, like the tutorial we’ll store the models in memory for now.
views.py
"""Cornice services."""
from cornice import Service
_TASKS = {}
tasks = Service(name='tasks', path='/tasks', description="Tasks")
@tasks.get()
def get_info(request):
"""Returns a list of all tasks."""
return {'tasks': _TASKS.keys()}
@tasks.post()
def create_task(request):
"""Adds a new task."""
task = request.json
if task['name'] in _TASKS:
raise Exception('That task already exists!')
_TASKS[task['name']] = task
and if we make a script to exercise it like this:
import simplejson as json
import requests
task = {
'name': 'take_out_the_trash',
'description': ("empty the trashcan and put the bag in the outside trashcan, "
"don't forget to put a new bag in!"),
}
response = requests.get('http://localhost:6543/tasks')
print response.status_code, response.text
response = requests.post('http://localhost:6543/tasks', json.dumps(task))
print response.status_code, response.text
response = requests.get('http://localhost:6543/tasks')
print response.status_code, response.text
it gives output:
200 {"tasks": []}
200 null
200 {"tasks": ["take_out_the_trash"]}
All right! we’re RESTing!
Hook up to a database with SQLAlchemy
We need to:
- Define a SQLAlchemy model.
- Create a database and create the table structure from the models.
- Set up connections to the database when the webapp loads.
- Add some config to define the url for the database and ask pyramid to handle transaction management for us.
1. models.py
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
from zope.sqlalchemy import ZopeTransactionExtension
Base = declarative_base()
DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
class Task(Base):
<strong>tablename</strong> = 'task'
task_id = Column(Integer, primary_key=True)
name = Column(Text, nullable=False)
description = Column(Text)
@classmethod
def from_json(cls, data):
return cls(**data)
def to_json(self):
to_serialize = ['task_id', 'name', 'description']
d = {}
for attr_name in to_serialize:
d[attr_name] = getattr(self, attr_name)
return d
2. Create a database and create the tables from our models.
I’m using Postgres so I run:
$ createdb blogpostcorniceapp
and then create a script to initialize the db:
from sqlalchemy import engine_from_config
from blogpostcorniceapp.models import DBSession
from pyramid.paster import get_appsettings
from blogpostcorniceapp.models import Base
settings = get_appsettings('/home/fran/blogpostcorniceapp/blogpostcorniceapp.ini')
engine = engine_from_config(settings, 'sqlalchemy.')
DBSession.configure(bind=engine)
Base.metadata.create_all(engine)
and now when I want to wipe the db and start again I can run:
$ dropdb blogpostcorniceapp && createdb blogpostcorniceapp && python initialize_db.py
3. main()
In blogpostcorniceapp/init.py in main() we want to add:
engine = engine_from_config(settings, 'sqlalchemy.')
DBSession.configure(bind=engine)
which will require these imports:
from sqlalchemy import engine_from_config
from blogpostcorniceapp.models import DBSession
and that will create connections to the database for the webapp.
4. blogpostcorniceapp.ini
Lastly we need two bits of configuration to define where the database can be found, and to ask pyramid to manage database transactions for us (so the transaction will be committed after the web request is complete or rolled back on errors without us having to do anything) in blogpostcorniceapp.ini
in the [app:main]
section:
pyramid.includes = pyramid_tm
sqlalchemy.url = postgres://fran@localhost/blogpostcorniceapp
and now we can change our views.py
to look like this:
@tasks.get()
def get_info(request):
"""Returns a list of all tasks."""
return {'tasks': [task.name for task in DBSession.query(Task)]}
@tasks.post()
def create_task(request):
"""Adds a new task."""
task = request.json
num_existing = DBSession.query(Task).filter(Task.name==task['name']).count()
if num_existing > 0:
raise Exception('That task already exists!')
DBSession.add(Task.from_json(task))
and running the script to exercise the app we get output:
200 {"tasks": []}
200 null
200 {"tasks": ["take_out_the_trash"]}
nice!
A little bit more Cornice
Cornice has a better way to model a RESTful API on a collection of models. We should be using a resource instead of a service. So we’ll rewrite the views.py like this:
from cornice.resource import resource
from cornice.resource import view
from blogpostcorniceapp.models import Task
from blogpostcorniceapp.models import DBSession
@resource(collection_path='/tasks', path='/tasks/{id}')
class TaskResource(object):
def __init__(self, request):
self.request = request
def collection_get(self):
return {'tasks': [task.name for task in DBSession.query(Task)]}
def collection_post(self):
"""Adds a new task."""
task = self.request.json
num_existing = DBSession.query(Task).filter(Task.name==task['name']).count()
if num_existing > 0:
raise Exception('That task already exists!')
DBSession.add(Task.from_json(task))
and the exercise script still gets the output we expect:
200 {"tasks": []}
200 null
200 {"tasks": ["take_out_the_trash"]}
A method on a class decorated with resource named for a HTTP verb (get, put, post, delete) will be exposed (in our case) at /tasks/{id}. A method called collection_put (or any HTTP verb) will be exposed at /tasks. To add a get for a individual task add to TaskResource:
def get(self):
return DBSession.query(Task).get(int(self.request.matchdict['id'])).to_json()
A Cornice Validator
Currently if we send two tasks with the same name:
response = requests.get('http://localhost:6543/tasks')
print response.status_code, response.text
response = requests.post('http://localhost:6543/tasks', json.dumps(task))
print response.status_code, response.text
response = requests.post('http://localhost:6543/tasks', json.dumps(task))
print response.status_code, response.text
we get:
200 {"tasks": []}
200 null
500 Internal Server Error
The server encountered an unexpected internal server error
(generated by waitress)
The unhandled Exception raised in TaskResource collection_post causes the app to generate a 500 Internal Server Error. Let’s make it so we get a useful error message back instead. In views.py, move the code that checks for other tasks with this name to a separate method and decorate collection_post to use that method as a validator:
@view(validators=('validate_post',))
def collection_post(self):
"""Adds a new task.""";
task = self.request.json
DBSession.add(Task.from_json(task))
def validate_post(self, request):
name = request.json['name']
num_existing = DBSession.query(Task).filter(Task.name==name).count()
if num_existing > 0:
request.errors.add(request.url, 'Non-unique task name.', 'There is already a task with this name.')
And now if we try to add the same task twice as above we get:
200 {"tasks": []}
200 null
400 {"status": "error", "errors": [{"location": "http://localhost:6543/tasks", "name": "Non-unique task name.", "description": "There is already a task with this name."}]}
So . . .
Cornice makes creating a RESTy webapp quite straightforward. And we showed how to integrate it with SQLAlchemy.
Hope this helps.