Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
from json import dumps, loads
from uuid import uuid4
from collections import OrderedDict
from klein import Klein
from exceptions import BaseException
class UniqueKeyError(BaseException):
pass
class NotFoundError(BaseException):
pass
class ValidationError(BaseException):
pass
class CustomerStore(object):
app = Klein()
def __init__(self):
self.customers = OrderedDict(dict()) # Trivial to swap out for some simple k/v store or SimpleDB
@app.route('/', methods=['GET', 'POST'])
def create_customer(self, request):
request.setHeader('Content-Type', 'application/json')
if request.method == 'POST':
body = loads(request.content.read())
if body.keys() != ['first_name', 'last_name', 'country']:
raise ValidationError('You must include `first_name` and `last_name` keys')
pk = '{0} {1}'.format(body['first_name'], body['last_name'])
if pk in self.customers:
raise UniqueKeyError('First name + last_name combination must be unique')
body['id'] = uuid4().get_hex()
self.customers[pk] = body
return dumps({'created': body})
else:
if not request.args or 'limit' in request.args:
if not self.customers:
raise NotFoundError('No customers in collection')
return dumps(self.customers[:request.args.get('limit', 2)]) # Show only two records by default
elif request.args.keys() != ['first_name', 'last_name']:
raise ValidationError('You must include `first_name` and `last_name` keys')
pk = '{0} {1}'.format(request.args['first_name'][0], request.args['last_name'][0])
if pk not in self.customers:
raise NotFoundError('First name + last_name combination not found in collection')
return dumps(self.customers[pk])
@app.route('/<string:name>', methods=['PUT'])
def save_customer(self, request, name):
request.setHeader('Content-Type', 'application/json')
body = loads(request.content.read())
# You can also edit the pk here, which might not be a good idea:
if {'first_name', 'last_name', 'id'}.issubset(set(body.keys())): # Allow you to edit `id` here
raise ValidationError('You must include `first_name`, `last_name`, `country` and/or `id` key(s)')
if name not in self.customers:
raise NotFoundError('"{0}" not found in customers collection'.format(name))
self.customers[name].update(body)
return dumps(self.customers[name])
@app.route('/<string:name>', methods=['GET'])
def retrieve_customer(self, request, name):
request.setHeader('Content-Type', 'application/json')
if name not in self.customers:
raise NotFoundError('"{0}" not found in customers collection'.format(name))
return dumps(self.customers[name])
@app.route('/<string:name>', methods=['DELETE'])
def delete_customer(self, request, name):
request.setHeader('Content-Type', 'application/json')
if name not in self.customers:
raise NotFoundError('"{0}" not found in customers collection'.format(name))
return dumps({'deleted': self.customers.pop(name)})
@app.handle_errors(UniqueKeyError, ValidationError)
def error_handler_for_unique_and_validation(self, request, failure):
request.setResponseCode(400)
print 'failure =', failure
print 'failure =', dir(failure)
return dumps({'error': '',
'error_message': '',
'stack': failure.getTraceback()})
@app.handle_errors(NotFoundError)
def error_handler_for_not_found(self, request, failure):
request.setResponseCode(404)
print 'failure =', failure
print 'failure =', dir(failure)
return dumps({'error': 'NotFound',
'error_message': '',
'stack': failure.getTraceback()})
if __name__ == '__main__':
store = CustomerStore()
@store.app.handle_errors
def error_handler_for_not_found(request, failure):
request.setResponseCode(404)
print 'failure =', failure
print 'failure =', dir(failure)
return dumps({'error': 'NotFound',
'error_message': '',
'stack': failure.getTraceback()})
store.app.run('0.0.0.0', 8080)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.