Let’s create a full working application with Cornice. We want to create a light messaging service.
You can find its whole source code at https://github.com/Cornices/examples/blob/master/messaging
Features:
Limitations:
The application provides two services:
On the server, the data is kept in memory.
We’ll provide a single CLI client in Python, using Curses.
Make sure you have virtualenv (see http://pypi.python.org/pypi/virtualenv).
Create a new directory and a virtualenv in it:
$ mkdir messaging
$ cd messaging
$ virtualenv --no-site-packages .
Once you have it, install Cornice in it with Pip:
$ bin/pip install cornice
We provide a Cookiecutter template you can use to create a new application:
$ bin/pip install cookiecutter
$ bin/cookiecutter gh:Cornices/cookiecutter-cornice
repo_name [myapp]: messaging
project_title [My Cornice application.]: Cornice tutorial
Once your application is generated, go there and call develop against it:
$ cd messaging
$ ../bin/python setup.py develop
...
The application can now be launched via embedded Pyramid pserve
, it provides a default “Hello”
service check:
$ ../bin/pserve messaging.ini
Starting server in PID 7618.
serving on 0.0.0.0:6543 view at http://127.0.0.1:6543
Once the application is running, visit http://127.0.0.1:6543 in your browser and make sure you get:
{'Hello': 'World'}
You should also get the same results calling the URL via Curl:
$ curl -i http://0.0.0.0:6543/
This will result:
HTTP/1.1 200 OK
Content-Length: 18
Content-Type: application/json; charset=UTF-8
Date: Tue, 12 May 2015 13:23:32 GMT
Server: waitress
{"Hello": "World"}
Let’s open the file in messaging/views.py
, it contains all the Services:
from cornice import Service
hello = Service(name='hello', path='/', description="Simplest app")
@hello.get()
def get_info(request):
"""Returns Hello in JSON."""
return {'Hello': 'World'}
We’re going to get rid of the Hello service, and change this file in order to add our first service - the users management
from cornice import Service
_USERS = {}
users = Service(name='users', path='/users', description="User registration")
@users.get(validators=valid_token)
def get_users(request):
"""Returns a list of all users."""
return {'users': _USERS.keys()}
@users.post(validators=unique)
def create_user(request):
"""Adds a new user."""
user = request.validated['user']
_USERS[user['name']] = user['token']
return {'token': '%s-%s' % (user['name'], user['token'])}
@users.delete(validators=valid_token)
def delete_user(request):
"""Removes the user."""
name = request.validated['user']
del _USERS[name]
return {'Goodbye': name}
What we have here is 3 methods on /users:
Remarks:
Validators are filling the request.validated mapping, the service can then use.
import os
import binascii
from pyramid.httpexceptions import HTTPUnauthorized
from cornice import Service
def _create_token():
return binascii.b2a_hex(os.urandom(20))
def valid_token(request):
header = 'X-Messaging-Token'
htoken = request.headers.get(header)
if htoken is None:
raise HTTPUnauthorized()
try:
user, token = htoken.split('-', 1)
except ValueError:
raise HTTPUnauthorized()
valid = user in _USERS and _USERS[user] == token
if not valid:
raise HTTPUnauthorized()
request.validated['user'] = user
def unique(request):
name = request.body
if name in _USERS:
request.errors.add('url', 'name', 'This user exists!')
else:
user = {'name': name, 'token': _create_token()}
request.validated['user'] = user
When the validator finds errors, it adds them to the request.errors mapping, and that will return a 400 with the errors.
Let’s try our application so far with CURL:
$ curl http://localhost:6543/users
{"status": 401, "message": "Unauthorized"}
$ curl -X POST http://localhost:6543/users -d 'tarek'
{"token": "tarek-a15fa2ea620aac8aad3e1b97a64200ed77dc7524"}
$ curl http://localhost:6543/users -H "X-Messaging-Token:tarek-a15fa2ea620aac8aad3e1b97a64200ed77dc7524"
{"users": ["tarek"]}
$ curl -X DELETE http://localhost:6543/users -H "X-Messaging-Token:tarek-a15fa2ea620aac8aad3e1b97a64200ed77dc7524"
{"Goodbye": "tarek"}
Now that we have users, let’s post and get messages. This is done via two very
simple functions we’re adding in the views.py
file:
_MESSAGES = []
messages = Service(name='messages', path='/', description="Messages")
@messages.get()
def get_messages(request):
"""Returns the 5 latest messages"""
return _MESSAGES[:5]
@messages.post(validators=(valid_token, valid_message))
def post_message(request):
"""Adds a message"""
_MESSAGES.insert(0, request.validated['message'])
return {'status': 'added'}
The first one simply returns the five first messages in a list, and the second one inserts a new message in the beginning of the list.
The POST uses two validators:
valid_token()
: the function we used previously that makes sure the
user is registeredvalid_message()
: a function that looks at the message provided in the
POST body, and puts it in the validated dict.Here’s the valid_message()
function:
import json
def valid_message(request):
try:
message = json.loads(request.body)
except ValueError:
request.errors.add('body', 'message', 'Not valid JSON')
return
# make sure we have the fields we want
if 'text' not in message:
request.errors.add('body', 'text', 'Missing text')
return
if 'color' in message and message['color'] not in ('red', 'black'):
request.errors.add('body', 'color', 'only red and black supported')
elif 'color' not in message:
message['color'] = 'black'
message['user'] = request.validated['user']
request.validated['message'] = message
This function extracts the json body, then checks that it contains a text key at least. It adds a color or use the one that was provided, and reuse the user name provided by the previous validator with the token control.
A simple client to use against our service can do three things:
Without going into great details, there’s a Python CLI against messaging that uses Curses.
See https://github.com/Cornices/examples/blob/master/messaging/messaging/client.py