Cornice and SQLAlchemy

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.

Cornice Validator

Cornice Validator

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:

[sourcecode language=”python”]
""" 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
[/sourcecode]

and if we make a script to exercise it like this:

[sourcecode language=”python”]
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
[/sourcecode]

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

[sourcecode language=”python”]
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):
__tablename__ = ‘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
[/sourcecode]

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:

[sourcecode language=”python”]
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)
[/sourcecode]

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:

[sourcecode language=”python”]
engine = engine_from_config(settings, ‘sqlalchemy.’)
DBSession.configure(bind=engine)
[/sourcecode]

which will require these imports:

[sourcecode language=”python”]
from sqlalchemy import engine_from_config
from blogpostcorniceapp.models import DBSession
[/sourcecode]

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:

[sourcecode language=”python”]
pyramid.includes = pyramid_tm
sqlalchemy.url = postgres://fran@localhost/blogpostcorniceapp
[/sourcecode]

and now we can change our views.py to look like this:

[sourcecode language=”python”]
@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))
[/sourcecode]

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:

[sourcecode language=”python”]
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))
[/sourcecode]

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:

[sourcecode language=”python”]
def get(self):
return DBSession.query(Task).get(int(self.request.matchdict[‘id’])).to_json()
[/sourcecode]

A Cornice Validator

Currently if we send two tasks with the same name:

[sourcecode language=”python”]
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
[/sourcecode]

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:

[sourcecode language=”python”]
@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.’)
[/sourcecode]

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.

Tags:

Creative Commons License

This work is licensed under a Creative Commons Attribution-NonCommercial-NoDerivatives 4.0 International License.

2 Comments

  1. Pyramid Climber

    very good tutorial, thank you, but this site has endless reloads of app.leadliaison.com, maybe you would like to fix that, it´s disturbing.

    • Brett g Porter

      Thanks — should be fixed now.