Created
December 5, 2008 16:53
-
-
Save galvez/32402 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This is some skeleton code to building a RESTful API using | |
web.py 0.3, my own xmlbuilder library and cElementTree for | |
generating and parsing XML, respectively; and Python's | |
low-level MySQLdb library. | |
It shows some conventions that I've adopted which are really | |
not very mature yet. It will eventually evolve into a well | |
defined set of standards and practices and SQLAlchemy is | |
also likely to be thrown into the mix in the future. | |
I'm running it in production today for a big API with nginx | |
load balancing requests to a few several instances of serve.py, | |
which will launch the default web.py threaded WSGI server. It | |
performs nicely (over 500 reqs/second) provided you avoid | |
long-running tasks in the request-response cycle (and instead | |
queue them in the background). | |
--Jonas Galvez, http://jonasgalvez.com.br/ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import MySQLdb | |
_DBs = { | |
'production': { | |
'host': '...', 'user': '...', 'passwd': '...', 'db': '...' | |
}, | |
'local': { | |
'host': 'localhost', 'user': '...', 'passwd': '...', 'db': '...' | |
} | |
} | |
# do export APP_ENV=value on your shell to change it | |
APP_ENV = os.environ.get('APP_ENV', 'local') | |
# bare-knuckles low-level DB approach | |
def DBCursor(): | |
conn = MySQLdb.connect(**_DBs[APP_ENV]) | |
cursor = conn.cursor(MySQLdb.cursors.DictCursor) | |
setattr(cursor, 'conn', conn) | |
return cursor |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from __future__ import with_statement | |
from xml.etree.cElementTree import iterparse | |
from xml.etree.ElementTree import ElementTree | |
from StringIO import StringIO | |
class PersonParser: | |
def __init__(self, xml=None): | |
self.name = None | |
self.uri = None | |
if xml != None: | |
self.fromstring(xml) | |
def fromstring(self, xml): | |
for event, elem in iterparse(StringIO(xml), events=('start', 'end')): | |
if event == 'start' and elem.tag == 'person': | |
self.uri = elem.get('uri') | |
elif event == 'start' and elem.tag == 'name': | |
self.name = elem.text | |
if __name__ == "__main__": | |
from xmlbuilder import builder | |
xml = builder(version="1.0", encoding="utf-8") | |
with xml.person(uri="/api/person/1"): | |
xml.name("FirstName LastName") | |
person = PersonParser(xml=str(xml)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from __future__ import with_statement | |
from xmlbuilder import builder | |
from dbcursor import DBCursor | |
from utils import * | |
from parsers.person import PersonParser | |
import re, web | |
class Person: | |
def GET(self, id): | |
db = DBCursor() | |
db.execute("""select * from persons where id = %s;""", (id,)) | |
person = db.fetchone() | |
if person is None: | |
web.ctx.status = '404 Not Found' | |
return '' | |
xml = builder(version="1.0", encoding="utf-8") | |
with xml.person(uri=uri.person(person['id'])): | |
xml.name(person['name']) | |
db.conn.close() | |
db.close() | |
response_xml = str(xml) | |
web.header('Content-Type', 'application/xml') | |
web.header('Content-Length', len(response_xml)) | |
return response_xml | |
def PUT(self): | |
person = PersonParser(xml=web.data()) | |
updated_fields = {} | |
db = DBCursor() | |
if person.name != None: | |
updated_fields.update({'name': person.name}) | |
if len(updated_fields.keys()) > 0: | |
sql = "update persons set %s" | |
sets = [] | |
for key in updated_fields.keys(): | |
sets.append("%s = %s" % (key, "%s")) | |
db.execute(sql % ", ".join(sets), tuple(updated_fields.values())) | |
db.conn.commit() | |
db.close() | |
db.conn.close() | |
web.ctx.status = '200 Ok' | |
web.header('Content-Location', uri.person(person_id)) | |
return | |
class PersonStore: | |
REQUIRED_FIELDS = ['name'] | |
def POST(self): | |
person = PersonParser(xml=web.data()) | |
for required_field in PersonStore.REQUIRED_FIELDS: | |
if (person.__dict__.get(required_field) == None): | |
web.ctx.status = '400 Bad Request' | |
return | |
db = DBCursor() | |
db.execute("insert into persons(name) values (%s);""", (person.name,)) | |
db.execute("""select last_insert_id() as person_id;""") | |
db.conn.commit() | |
person_id = db.fetchone()['person_id'] | |
db.close() | |
db.conn.close() | |
web.ctx.status = '201 Created' | |
web.header('Content-Location', uri.person(person_id)) | |
return |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from __future__ import with_statement | |
import sys, time, os.path, web | |
base = os.path.abspath(os.path.dirname(sys.argv[0])) | |
sys.path.append(os.path.join(base, os.path.pardir)) | |
sys.path.append(os.path.join(base, os.path.pardir, 'lib')) | |
URIs = ( | |
'/api/person', 'resources.person.PersonStore', # POST | |
'/api/person/(\d+)', 'resources.person.Person' # GET/PUT/DELETE | |
) | |
app = web.application(URIs, globals()) | |
if __name__ == "__main__": | |
app.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class uri: | |
@staticmethod | |
def person(id): | |
return "/api/person/%s" % id |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from __future__ import with_statement | |
from StringIO import StringIO | |
__all__ = ['__author__', '__license__', 'builder', 'element'] | |
__author__ = ('Jonas Galvez', 'jonas@codeazur.com.br', 'http://jonasgalvez.com.br') | |
__license__ = "GPL" | |
import sys | |
class builder: | |
def __init__(self, version, encoding): | |
self.document = StringIO() | |
self.document.write('<?xml version="%s" encoding="%s"?>\n' % (version, encoding)) | |
self.indentation = -2 | |
def __getattr__(self, name): | |
return element(name, self) | |
def __getitem__(self, name): | |
return element(name, self) | |
def __str__(self): | |
return self.document.getvalue() | |
def __unicode__(self): | |
return self.document.getvalue().decode("utf-8") | |
def write(self, line): | |
self.document.write('%s%s' % ((self.indentation * ' '), line)) | |
_dummy = {} | |
class element: | |
def __init__(self, name, builder): | |
self.name = name | |
self.builder = builder | |
def __enter__(self): | |
self.builder.indentation += 2 | |
if hasattr(self, 'attributes'): | |
self.builder.write('<%s %s>\n' % (self.name, self.serialized_attrs)) | |
else: | |
self.builder.write('<%s>\n' % self.name) | |
def __exit__(self, type, value, tb): | |
self.builder.write('</%s>\n' % self.name) | |
self.builder.indentation -= 2 | |
def __call__(self, value=_dummy, **kargs): | |
if len(kargs.keys()) > 0: | |
self.attributes = kargs | |
self.serialized_attrs = self.serialize_attrs(kargs) | |
if value == None: | |
self.builder.indentation += 2 | |
if hasattr(self, 'attributes'): | |
self.builder.write('<%s %s />\n' % (self.name, self.serialized_attrs)) | |
else: | |
self.builder.write('<%s />\n' % self.name) | |
self.builder.indentation -= 2 | |
elif value != _dummy: | |
self.builder.indentation += 2 | |
if hasattr(self, 'attributes'): | |
self.builder.write('<%s %s>%s</%s>\n' % (self.name, self.serialized_attrs, value, self.name)) | |
else: | |
self.builder.write('<%s>%s</%s>\n' % (self.name, value, self.name)) | |
self.builder.indentation -= 2 | |
return | |
return self | |
def serialize_attrs(self, attrs): | |
serialized = [] | |
for attr, value in attrs.items(): | |
serialized.append('%s="%s"' % (attr, value)) | |
return ' '.join(serialized) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment