I am working on a sample REST service to study testing related aspects using the Python Bottle framework. I read a lot about REST but I do not have a lot of practical experience designing RESTful APIs (e.g. this https://www.ics.uci.edu/~fielding/pubs/dissertation/fielding_dissertation.pdf). I am using Bottle the first time so there is definitely room for improvements.
I was looking for some complete examples using Bottle. These are the ones I could find:
- http://gotofritz.net/blog/weekly-challenge/restful-python-api-bottle/
- https://myadventuresincoding.wordpress.com/2011/01/02/creating-a-rest-api-in-python-using-bottle-and-mongodb/
- http://www.giantflyingsaucer.com/blog/?p=5469
These samples did not exactly cover what I had in mind so I started to work on a new one with the following goals in mind:
- learn about REST APIs
- understand the testing related aspects (using nosetests)
- sample that can be uses as a cookie cutter template to building more usefull RESTful services
- support Travis for continuous build
I am mainly looking to perfect the REST API and the testing related aspects. But any improvement be it coding style, structure etc. is welcome. Please note that deployment related aspects are out of scope for now.
restserver.py:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
This is a demo REST server. Do not use this in production!
"""
import json
from bottle import route, run, request, response, abort, default_app, static_file, hook
import os
__version__ = "0.1.1"
here = lambda x: os.path.abspath(os.path.join(os.path.dirname(__file__), x))
SUPERCARS_FILE = here('supercars.json')
supercars = {}
def load_supercars(filename):
"""Load supercars from json file"""
with open(filename) as data_file:
data = json.load(data_file)
for oid, entry in enumerate(data):
supercars[oid] = entry
return supercars
@hook('after_request')
def enable_cors():
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With'
@route('/supercars/<id:int>', method='GET')
def supercar(id):
"""Respond to a GET request."""
# curl -X GET 'http://localhost:8080/supercars/1'
try:
return supercars[int(id)]
except KeyError:
abort(404, 'No supercar with id %s' % id)
@route('/supercars', method='OPTIONS')
def cors_support():
return {}
@route('/supercars', method='GET')
def all_supercars():
"""Respond to a GET request."""
# curl -X GET 'http://localhost:8080/supercars/'
# need to return dict: http://stackoverflow.com/questions/12293979/how-do-i-return-a-json-array-with-bottle
response.content_type = 'application/json'
return json.dumps(supercars.values())
@route('/supercars/<id:int>', method='DELETE')
def delete_supercar(id):
"""Respond to a DELETE request."""
# curl -X DELETE 'http://localhost:8080/supercars/1'
try:
supercars.pop(int(id))
except KeyError:
abort(404, 'No supercar with id %s' % id)
@route('/supercars/<id:int>', method='PUT')
def update_supercar(id):
"""Respond to a PUT request."""
# curl -X PUT -d '{"have_one":"yes"}' 'http://localhost:8080/supercars/1'
if not request.json:
abort(400, 'No data received')
car = request.json
try:
u = supercars[int(id)].copy()
u.update(car)
supercars[int(id)] = u
except KeyError:
abort(404, 'No supercar with id %s' % id)
@route('/supercars', method='POST')
def insert_supercar():
"""Respond to a POST request."""
# curl -X POST -d '{"name":"Ferrari Enzo","country":"Italy","top_speed":"218","0-60":"3.4","power":"650","engine":"5998","weight":"1365","description":"The Enzo Ferrari is a 12 cylinder mid-engine berlinetta named after the company\"s founder, Enzo Ferrari.","image":"050.png"}' 'http://localhost:8080/supercars'
data = request.body.readline()
if not data:
abort(400, 'No data received')
car = json.loads(data)
seq = max(supercars.keys()) + 1
supercars[seq] = car
@route('/images/<filename>', method='GET')
def server_static(filename):
"""Serve static image file."""
return static_file(filename, root=here('../images'))
if __name__ == '__main__':
supercars = load_supercars(SUPERCARS_FILE)
# remember to remove reloader=True and debug=True when you move your application from development to next stages
run(host='localhost', port=8080, debug=True, reloader=True)
tests/test_restserver.py:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from supercars.restserver import supercars, load_supercars, supercar, all_supercars, delete_supercar, update_supercar, \
insert_supercar
from bottle import HTTPError, request, tob
from nose.tools import with_setup, assert_equal, raises
import tempfile
from io import BytesIO
def fake_request_json(body):
"""helper function to set the fake request body."""
request.environ['CONTENT_LENGTH'] = str(len(tob(body)))
request.environ['CONTENT_TYPE'] = 'application/json'
if 'wsgi.input' not in request.environ:
request.environ['wsgi.input'] = BytesIO()
request.environ['wsgi.input'].seek(0)
request.environ['wsgi.input'].write(tob(body))
request.environ['wsgi.input'].seek(0)
def clear_supercars():
"""test setup clears the supercars dictionary."""
supercars.clear()
@with_setup(clear_supercars)
def test_supercar():
supercars[1] = {'engine': '1'}
assert_equal(supercar('1'), {'engine': '1'})
@raises(HTTPError)
@with_setup(clear_supercars)
def test_supercar_notfound():
supercars[1] = {'engine': '1'}
supercar('2')
@with_setup(clear_supercars)
def test_load_supercars():
temp = tempfile.NamedTemporaryFile()
try:
temp.write('[{"json": "true"}]')
temp.seek(0)
result = load_supercars(temp.name)
assert_equal(result, {0: {'json': 'true'}})
finally:
temp.close()
@raises(IOError)
def test_load_supercars_nosuchfile():
load_supercars('nosuchfile.txt')
@raises(ValueError)
def test_load_supercars_nojson():
temp = tempfile.NamedTemporaryFile()
try:
temp.write('no json content')
temp.seek(0)
load_supercars(temp.name)
finally:
temp.close()
@with_setup(clear_supercars)
def test_all_supercars():
supercars[1] = {'engine': '1'}
supercars[2] = {'engine': '2'}
assert_equal(all_supercars(), {1: {'engine': '1'}, 2: {'engine': '2'}})
@with_setup(clear_supercars)
def test_all_supercars_empty():
assert_equal(all_supercars(), {})
@with_setup(clear_supercars)
def test_delete_supercar():
supercars[1] = {'engine': '1'}
delete_supercar(1)
assert_equal(all_supercars(), {})
@raises(HTTPError)
@with_setup(clear_supercars)
def test_delete_supercar_notfound():
delete_supercar(1)
@with_setup(clear_supercars)
def test_update_supercar():
supercars[1] = {'engine': '1'}
fake_request_json('{"have_one": "yes"}')
update_supercar(1)
assert_equal(supercar(1), {'engine': '1', 'have_one': 'yes'})
@raises(HTTPError)
@with_setup(clear_supercars)
def test_update_supercar_unknown():
supercars[1] = {'engine': '1'}
update_supercar(2)
@with_setup(clear_supercars)
def test_insert_supercar():
supercars[1] = {'engine': 1}
fake_request_json('{"engine": "wwwrrrrmmmm"}')
insert_supercar()
assert_equal(all_supercars(), {1: {'engine': 1}, 2: {'engine': 'wwwrrrrmmmm'}})
and the test for the routes: tests/test_routes.py:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from supercars.restserver import supercars, default_app
from supercars import restserver
from nose.tools import with_setup, assert_equal
from webtest import TestApp
import json
def clear_supercars():
"""test setup clears the supercars dictionary."""
supercars.clear()
@with_setup(clear_supercars)
def test_get_supercar():
supercars[1] = {'_id': '1'}
app = TestApp(default_app())
resp = app.get('/supercars/1')
assert_equal(resp.status, '200 OK')
assert_equal(resp.json, {'_id': '1'})
@with_setup(clear_supercars)
def test_get_all_supercars():
supercars[1] = {'engine': '1'}
app = TestApp(default_app())
resp = app.get('/supercars')
assert_equal(resp.status, '200 OK')
assert_equal(resp.json, {'1': {'engine': '1'}})
@with_setup(clear_supercars)
def test_delete_supercar():
supercars[1] = {'engine': '1'}
app = TestApp(default_app())
resp = app.delete('/supercars/1')
assert_equal(resp.status, '200 OK')
app.reset()
@with_setup(clear_supercars)
def test_update_supercar():
supercars[1] = {'engine': '1'}
app = TestApp(restserver.default_app())
# this MUST be json data!
# resp = app.put('/supercars/1', json.dumps({'have_one': 'yes'}), content_type='application/json')
# resp = app.put_json('/supercars/1', {'have_one': 'yes'})
resp = app.put('/supercars/1', '{"have_one": \n"yes"}', content_type='application/json')
assert_equal(resp.status, '200 OK')
assert_equal(supercars, {1: {'engine': '1', 'have_one': 'yes'}})
@with_setup(clear_supercars)
def test_insert_supercar():
supercars[1] = {'engine': '1'}
app = TestApp(restserver.default_app())
resp = app.post_json('/supercars', {'engine': '2'})
assert_equal(resp.status, '200 OK')
assert_equal(supercars, {1: {'engine': '1'}, 2: {'engine': '2'}})
def test_get_image():
app = TestApp(restserver.default_app())
resp = app.get('/images/005.png')
assert_equal(resp.status, '200 OK')
assert_equal(resp.content_length, 115616)
