Last active
October 24, 2015 15:48
-
-
Save davidbgk/e459577c241e5b905571 to your computer and use it in GitHub Desktop.
Falcon-RESTQL: POC of an on-demand field serialization using REST à la GraphQL
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
# Virtualenv: python3.5 -m venv ~/.virtualenvs/falcon-restql | |
# Activate: source ~/.virtualenvs/falcon-restql/bin/activate | |
# Install: pip3.5 install --upgrade cython falcon redis-limpyd httpie gunicorn | |
# Launch: gunicorn falcon-restql:app --reload + redis-server | |
# Test: | |
# $ http :8000/things/bicycle | |
# => [] | |
# $ http POST :8000/things/bicycle name="bicycle" color="red" | |
# => location: /things/bicycle | |
# $ http :8000/things/bicycle | |
# => {"color": "red", "name": "bicycle"} | |
# $ http :8000/things/bicycle X-Fields:color | |
# => {"color": "red"} | |
import json | |
import falcon | |
from limpyd import model | |
main_database = model.RedisDatabase( | |
host="localhost", | |
port=6379, | |
db=1 | |
) | |
class ThingModel(model.RedisModel): | |
database = main_database | |
name = model.InstanceHashField(indexable=True, unique=True) | |
color = model.InstanceHashField() | |
class StorageEngine: | |
def get_things(self, name): | |
return ThingModel.collection(name=name).instances()[0] | |
def add_thing(self, name, color): | |
thing = ThingModel(name=name) | |
thing.hmset(color=color) | |
return thing | |
class StorageError(Exception): | |
@staticmethod | |
def handle(ex, req, resp, params): | |
description = ('Sorry, couldn\'t write your thing to the ' | |
'database. It worked on my box.') | |
raise falcon.HTTPError(falcon.HTTP_725, | |
'Database Error', | |
description) | |
class RequireJSON: | |
def process_request(self, req, resp): | |
if not req.client_accepts_json: | |
raise falcon.HTTPNotAcceptable( | |
'This API only supports responses encoded as JSON.', | |
href='http://docs.examples.com/api/json') | |
if req.method in ('POST', 'PUT'): | |
if 'application/json' not in req.content_type: | |
raise falcon.HTTPUnsupportedMediaType( | |
'This API only supports requests encoded as JSON.', | |
href='http://docs.examples.com/api/json') | |
class JSONTranslator: | |
def process_request(self, req, resp): | |
if req.content_length in (None, 0): | |
# Nothing to do | |
return | |
body = req.stream.read() | |
if not body: | |
raise falcon.HTTPBadRequest('Empty request body', | |
'A valid JSON document is required.') | |
try: | |
req.context['doc'] = json.loads(body.decode('utf-8')) | |
except (ValueError, UnicodeDecodeError): | |
raise falcon.HTTPError(falcon.HTTP_753, | |
'Malformed JSON', | |
'Could not decode the request body. The ' | |
'JSON was incorrect or not encoded as ' | |
'UTF-8.') | |
def process_response(self, req, resp, resource): | |
if 'result' not in req.context: | |
return | |
resp.body = json.dumps(req.context['result']) | |
class ThingsResource: | |
def __init__(self, db): | |
self.db = db | |
def on_get(self, req, resp, thing_name): | |
try: | |
result = self.db.get_things(thing_name) | |
except IndexError: | |
req.context['result'] = [] | |
return | |
except Exception: | |
description = ('Aliens have attacked our base! We will ' | |
'be back as soon as we fight them off. ' | |
'We appreciate your patience.') | |
raise falcon.HTTPServiceUnavailable( | |
'Service Outage', | |
description, | |
30) | |
keys = set(['color', 'name']) | |
# This is were the filtering happens given the X-Fields header. | |
if req.get_header('X-Fields'): | |
keys &= set(req.get_header('X-Fields').split(',')) | |
values = result.hmget(*keys) | |
req.context['result'] = dict(zip(keys, values)) | |
resp.status = falcon.HTTP_200 | |
def on_post(self, req, resp, thing_name): | |
try: | |
doc = req.context['doc'] | |
except KeyError: | |
raise falcon.HTTPBadRequest( | |
'Missing thing', | |
'A thing must be submitted in the request body.') | |
self.db.add_thing(doc['name'], doc['color']) | |
resp.status = falcon.HTTP_201 | |
resp.location = '/things/{name}'.format(name=thing_name) | |
app = falcon.API(middleware=[ | |
RequireJSON(), | |
JSONTranslator(), | |
]) | |
db = StorageEngine() | |
things = ThingsResource(db) | |
app.add_route('/things/{thing_name}', things) | |
app.add_error_handler(StorageError, StorageError.handle) |
Nice! So you chose the X-Fields approach with flat list ?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Next step: pass a
X-GraphQL
header with a GraphQL query and filter using Graphene or a more minimalist implementation.