openstack/deb-python-falcon
Code Issues Proposed changes

doc(guide): Update quickstart example code

Browse Source 
Update the README and quickstart example code to showcase the new
middleware feature. Also tweak other parts of the example code,
as needed, to be in sync with the latest updates to the framework.

Partially implements #296
This commit is contained in:
Kurt Griffiths
2014-12-24 18:25:47 -06:00
parent 6ff79451c2
commit 87236e8c1c
5 changed files with 460 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
README.md
View File

@@ -169,24 +169,27 @@ $ curl localhost:8000/things
Here is a more involved example that demonstrates reading headers and query parameters, handling errors, and working with request and response bodies.
```python
import json
import logging
import uuid
from wsgiref import simple_server
import falcon
import requests
class StorageEngine(object):
def get_things(self, marker, limit):
return []
return [{'id': str(uuid.uuid4()), 'color': 'green'}]
def add_thing(self, thing):
return {'id': str(uuid.uuid4())}
thing['id'] = str(uuid.uuid4())
return thing
class StorageError(Exception):
@staticmethod
def handle(ex, req, resp, params):
description = ('Sorry, couldn\'t write your thing to the '
@@ -197,82 +200,110 @@ class StorageError(Exception):
description)
class Proxy(object):
def forward(self, req):
return falcon.HTTP_503
class SinkAdapter(object):
def __init__(self):
self._proxy = Proxy()
engines = {
'ddg': 'https://duckduckgo.com',
'y': 'https://search.yahoo.com/search',
}
def __call__(self, req, resp, **kwargs):
resp.status = self._proxy.forward(req)
self.kwargs = kwargs
def __call__(self, req, resp, engine):
url = self.engines[engine]
params = {'q': req.get_param('q', True)}
result = requests.get(url, params=params)
resp.status = str(result.status_code) + ' ' + result.reason
resp.content_type = result.headers['content-type']
resp.body = result.text
def token_is_valid(token, user_id):
return True # Suuuuuure it's valid...
class AuthMiddleware(object):
def process_request(self, req, resp):
token = req.get_header('X-Auth-Token')
project = req.get_header('X-Project-ID')
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not self._token_is_valid(token, project):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def _token_is_valid(self, token, project):
return True # Suuuuuure it's valid...
def auth(req, resp, params):
# Alternatively, use Talons or do this in WSGI middleware...
token = req.get_header('X-Auth-Token')
class RequireJSON(object):
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not token_is_valid(token, params['user_id']):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def check_media_type(req, resp, params):
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 not req.content_type == 'application/json':
raise falcon.HTTPUnsupportedMediaType(
'This API only supports requests encoded as JSON.',
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')
def deserialize(req, resp, resource, params):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
body = req.stream.read()
if not body:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
try:
params['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.')
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')
def serialize(req, resp, resource):
resp.body = json.dumps(req.context['doc'])
class JSONTranslator(object):
def process_request(self, req, resp):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
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):
if 'result' not in req.context:
return
resp.body = json.dumps(req.context['result'])
def max_body(limit):
def hook(req, resp, resource, params):
length = req.content_length
if length is not None and length > limit:
msg = ('The size of the request is too large. The body must not '
'exceed ' + str(limit) + ' bytes in length.')
raise falcon.HTTPRequestEntityTooLarge(
'Request body is too large', msg)
return hook
class ThingsResource:
@@ -281,7 +312,6 @@ class ThingsResource:
self.db = db
self.logger = logging.getLogger('thingsapp.' + __name__)
@falcon.after(serialize)
def on_get(self, req, resp, user_id):
marker = req.get_param('marker') or ''
limit = req.get_param_as_int('limit') or 50
@@ -304,13 +334,20 @@ class ThingsResource:
# create a custom class that inherits from falcon.Request. This
# class could, for example, have an additional 'doc' property
# that would serialize to JSON under the covers.
req.context['doc'] = result
req.context['result'] = result
resp.set_header('X-Powered-By', 'Small Furry Creatures')
resp.status = falcon.HTTP_200
@falcon.before(deserialize)
def on_post(self, req, resp, user_id, doc):
@falcon.before(max_body(64 * 1024))
def on_post(self, req, resp, user_id):
try:
doc = req.context['doc']
except KeyError:
raise falcon.HTTPBadRequest(
'Missing thing',
'A thing must be submitted in the request body.')
proper_thing = self.db.add_thing(doc)
resp.status = falcon.HTTP_201
@@ -318,7 +355,11 @@ class ThingsResource:
# Configure your WSGI server to load "things.app" (app is a WSGI callable)
app = falcon.API(before=[auth, check_media_type])
app = falcon.API(middleware=[
AuthMiddleware(),
RequireJSON(),
JSONTranslator(),
])
db = StorageEngine()
things = ThingsResource(db)
@@ -332,13 +373,12 @@ app.add_error_handler(StorageError, StorageError.handle)
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter()
app.add_sink(sink, r'/v1/[charts|inventory]')
app.add_sink(sink, r'/search/(?P<engine>ddg|y)\Z')
# Useful for debugging problems in your API; works with pdb.set_trace()
if __name__ == '__main__':
httpd = simple_server.make_server('127.0.0.1', 8000, app)
httpd.serve_forever()
```
### Contributing ###

186
README.rst
View File

@@ -192,17 +192,21 @@ Here is a more involved example that demonstrates reading headers and query para
from wsgiref import simple_server
import falcon
import requests
class StorageEngine(object):
def get_things(self, marker, limit):
return []
return [{'id': str(uuid.uuid4()), 'color': 'green'}]
def add_thing(self, thing):
return {'id': str(uuid.uuid4())}
thing['id'] = str(uuid.uuid4())
return thing
class StorageError(Exception):
@staticmethod
def handle(ex, req, resp, params):
description = ('Sorry, couldn\'t write your thing to the '
@@ -213,82 +217,110 @@ Here is a more involved example that demonstrates reading headers and query para
description)
class Proxy(object):
def forward(self, req):
return falcon.HTTP_503
class SinkAdapter(object):
def __init__(self):
self._proxy = Proxy()
engines = {
'ddg': 'https://duckduckgo.com',
'y': 'https://search.yahoo.com/search',
}
def __call__(self, req, resp, **kwargs):
resp.status = self._proxy.forward(req)
self.kwargs = kwargs
def __call__(self, req, resp, engine):
url = self.engines[engine]
params = {'q': req.get_param('q', True)}
result = requests.get(url, params=params)
resp.status = str(result.status_code) + ' ' + result.reason
resp.content_type = result.headers['content-type']
resp.body = result.text
def token_is_valid(token, user_id):
return True # Suuuuuure it's valid...
class AuthMiddleware(object):
def process_request(self, req, resp):
token = req.get_header('X-Auth-Token')
project = req.get_header('X-Project-ID')
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not self._token_is_valid(token, project):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def _token_is_valid(self, token, project):
return True # Suuuuuure it's valid...
def auth(req, resp, params):
# Alternatively, use Talons or do this in WSGI middleware...
token = req.get_header('X-Auth-Token')
class RequireJSON(object):
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not token_is_valid(token, params['user_id']):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def check_media_type(req, resp, params):
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 not req.content_type == 'application/json':
raise falcon.HTTPUnsupportedMediaType(
'This API only supports requests encoded as JSON.',
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')
def deserialize(req, resp, resource, params):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
body = req.stream.read()
if not body:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
try:
params['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.')
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')
def serialize(req, resp, resource):
resp.body = json.dumps(req.context['doc'])
class JSONTranslator(object):
def process_request(self, req, resp):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
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):
if 'result' not in req.context:
return
resp.body = json.dumps(req.context['result'])
def max_body(limit):
def hook(req, resp, resource, params):
length = req.content_length
if length is not None and length > limit:
msg = ('The size of the request is too large. The body must not '
'exceed ' + str(limit) + ' bytes in length.')
raise falcon.HTTPRequestEntityTooLarge(
'Request body is too large', msg)
return hook
class ThingsResource:
@@ -297,7 +329,6 @@ Here is a more involved example that demonstrates reading headers and query para
self.db = db
self.logger = logging.getLogger('thingsapp.' + __name__)
@falcon.after(serialize)
def on_get(self, req, resp, user_id):
marker = req.get_param('marker') or ''
limit = req.get_param_as_int('limit') or 50
@@ -320,13 +351,20 @@ Here is a more involved example that demonstrates reading headers and query para
# create a custom class that inherits from falcon.Request. This
# class could, for example, have an additional 'doc' property
# that would serialize to JSON under the covers.
req.context['doc'] = result
req.context['result'] = result
resp.set_header('X-Powered-By', 'Small Furry Creatures')
resp.status = falcon.HTTP_200
@falcon.before(deserialize)
def on_post(self, req, resp, user_id, doc):
@falcon.before(max_body(64 * 1024))
def on_post(self, req, resp, user_id):
try:
doc = req.context['doc']
except KeyError:
raise falcon.HTTPBadRequest(
'Missing thing',
'A thing must be submitted in the request body.')
proper_thing = self.db.add_thing(doc)
resp.status = falcon.HTTP_201
@@ -334,7 +372,11 @@ Here is a more involved example that demonstrates reading headers and query para
# Configure your WSGI server to load "things.app" (app is a WSGI callable)
app = falcon.API(before=[auth, check_media_type])
app = falcon.API(middleware=[
AuthMiddleware(),
RequireJSON(),
JSONTranslator(),
])
db = StorageEngine()
things = ThingsResource(db)
@@ -348,7 +390,7 @@ Here is a more involved example that demonstrates reading headers and query para
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter()
app.add_sink(sink, r'/v1/[charts|inventory]')
app.add_sink(sink, r'/search/(?P<engine>ddg|y)\Z')
# Useful for debugging problems in your API; works with pdb.set_trace()
if __name__ == '__main__':

190
doc/user/quickstart.rst
View File

@@ -75,17 +75,21 @@ parameters, handling errors, and working with request and response bodies.
from wsgiref import simple_server
import falcon
import requests
class StorageEngine(object):
def get_things(self, marker, limit):
return []
return [{'id': str(uuid.uuid4()), 'color': 'green'}]
def add_thing(self, thing):
return {'id': str(uuid.uuid4())}
thing['id'] = str(uuid.uuid4())
return thing
class StorageError(Exception):
@staticmethod
def handle(ex, req, resp, params):
description = ('Sorry, couldn\'t write your thing to the '
@@ -96,82 +100,110 @@ parameters, handling errors, and working with request and response bodies.
description)
class Proxy(object):
def forward(self, req):
return falcon.HTTP_503
class SinkAdapter(object):
def __init__(self):
self._proxy = Proxy()
engines = {
'ddg': 'https://duckduckgo.com',
'y': 'https://search.yahoo.com/search',
}
def __call__(self, req, resp, **kwargs):
resp.status = self._proxy.forward(req)
self.kwargs = kwargs
def __call__(self, req, resp, engine):
url = self.engines[engine]
params = {'q': req.get_param('q', True)}
result = requests.get(url, params=params)
resp.status = str(result.status_code) + ' ' + result.reason
resp.content_type = result.headers['content-type']
resp.body = result.text
def token_is_valid(token, user_id):
return True # Suuuuuure it's valid...
class AuthMiddleware(object):
def process_request(self, req, resp):
token = req.get_header('X-Auth-Token')
project = req.get_header('X-Project-ID')
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not self._token_is_valid(token, project):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def _token_is_valid(self, token, project):
return True # Suuuuuure it's valid...
def auth(req, resp, params):
# Alternatively, use Talons or do this in WSGI middleware...
token = req.get_header('X-Auth-Token')
class RequireJSON(object):
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not token_is_valid(token, params['user_id']):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def check_media_type(req, resp, params):
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 not req.content_type == 'application/json':
raise falcon.HTTPUnsupportedMediaType(
'This API only supports requests encoded as JSON.',
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')
def deserialize(req, resp, resource, params):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
body = req.stream.read()
if not body:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
try:
params['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.')
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')
def serialize(req, resp, resource):
resp.body = json.dumps(req.context['doc'])
class JSONTranslator(object):
def process_request(self, req, resp):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
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):
if 'result' not in req.context:
return
resp.body = json.dumps(req.context['result'])
def max_body(limit):
def hook(req, resp, resource, params):
length = req.content_length
if length is not None and length > limit:
msg = ('The size of the request is too large. The body must not '
'exceed ' + str(limit) + ' bytes in length.')
raise falcon.HTTPRequestEntityTooLarge(
'Request body is too large', msg)
return hook
class ThingsResource:
@@ -180,7 +212,6 @@ parameters, handling errors, and working with request and response bodies.
self.db = db
self.logger = logging.getLogger('thingsapp.' + __name__)
@falcon.after(serialize)
def on_get(self, req, resp, user_id):
marker = req.get_param('marker') or ''
limit = req.get_param_as_int('limit') or 50
@@ -203,13 +234,20 @@ parameters, handling errors, and working with request and response bodies.
# create a custom class that inherits from falcon.Request. This
# class could, for example, have an additional 'doc' property
# that would serialize to JSON under the covers.
req.context['doc'] = result
req.context['result'] = result
resp.set_header('X-Powered-By', 'Small Furry Creatures')
resp.status = falcon.HTTP_200
@falcon.before(deserialize)
def on_post(self, req, resp, user_id, doc):
@falcon.before(max_body(64 * 1024))
def on_post(self, req, resp, user_id):
try:
doc = req.context['doc']
except KeyError:
raise falcon.HTTPBadRequest(
'Missing thing',
'A thing must be submitted in the request body.')
proper_thing = self.db.add_thing(doc)
resp.status = falcon.HTTP_201
@@ -217,7 +255,11 @@ parameters, handling errors, and working with request and response bodies.
# Configure your WSGI server to load "things.app" (app is a WSGI callable)
app = falcon.API(before=[auth, check_media_type])
app = falcon.API(middleware=[
AuthMiddleware(),
RequireJSON(),
JSONTranslator(),
])
db = StorageEngine()
things = ThingsResource(db)
@@ -231,13 +273,9 @@ parameters, handling errors, and working with request and response bodies.
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter()
app.add_sink(sink, r'/v1/[charts|inventory]')
app.add_sink(sink, r'/search/(?P<engine>ddg|y)\Z')
# Useful for debugging problems in your API; works with pdb.set_trace()
if __name__ == '__main__':
httpd = simple_server.make_server('127.0.0.1', 8000, app)
httpd.serve_forever()

8
doc/user/tutorial.rst
View File

@@ -3,10 +3,10 @@
Tutorial
========
This page walks you through building an API for an image-sharing service. Along
the way, you will learn about Falcon's features and the terminology used by
the framework. You'll also learn how to query Falcon's docstrings, and get a
quick overview of the WSGI standard.
This page walks you through building an API for a simple image-sharing
service. Along the way, you will learn about Falcon's features and the
terminology used by the framework. You'll also learn how to query Falcon's
docstrings, and get a quick overview of the WSGI standard.
.. include:: big-picture-snip.rst

186
tests/test_example.py
View File

@@ -4,17 +4,21 @@ import uuid
from wsgiref import simple_server
import falcon
import requests
class StorageEngine(object):
def get_things(self, marker, limit):
return []
return [{'id': str(uuid.uuid4()), 'color': 'green'}]
def add_thing(self, thing):
return {'id': str(uuid.uuid4())}
thing['id'] = str(uuid.uuid4())
return thing
class StorageError(Exception):
@staticmethod
def handle(ex, req, resp, params):
description = ('Sorry, couldn\'t write your thing to the '
@@ -25,82 +29,110 @@ class StorageError(Exception):
description)
class Proxy(object):
def forward(self, req):
return falcon.HTTP_503
class SinkAdapter(object):
def __init__(self):
self._proxy = Proxy()
engines = {
'ddg': 'https://duckduckgo.com',
'y': 'https://search.yahoo.com/search',
}
def __call__(self, req, resp, **kwargs):
resp.status = self._proxy.forward(req)
self.kwargs = kwargs
def __call__(self, req, resp, engine):
url = self.engines[engine]
params = {'q': req.get_param('q', True)}
result = requests.get(url, params=params)
resp.status = str(result.status_code) + ' ' + result.reason
resp.content_type = result.headers['content-type']
resp.body = result.text
def token_is_valid(token, user_id):
return True # Suuuuuure it's valid...
class AuthMiddleware(object):
def process_request(self, req, resp):
token = req.get_header('X-Auth-Token')
project = req.get_header('X-Project-ID')
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not self._token_is_valid(token, project):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def _token_is_valid(self, token, project):
return True # Suuuuuure it's valid...
def auth(req, resp, params):
# Alternatively, use Talons or do this in WSGI middleware...
token = req.get_header('X-Auth-Token')
class RequireJSON(object):
if token is None:
description = ('Please provide an auth token '
'as part of the request.')
raise falcon.HTTPUnauthorized('Auth token required',
description,
href='http://docs.example.com/auth')
if not token_is_valid(token, params['user_id']):
description = ('The provided auth token is not valid. '
'Please request a new token and try again.')
raise falcon.HTTPUnauthorized('Authentication required',
description,
href='http://docs.example.com/auth',
scheme='Token; UUID')
def check_media_type(req, resp, params):
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 not req.content_type == 'application/json':
raise falcon.HTTPUnsupportedMediaType(
'This API only supports requests encoded as JSON.',
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')
def deserialize(req, resp, resource, params):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
body = req.stream.read()
if not body:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
try:
params['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.')
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')
def serialize(req, resp, resource):
resp.body = json.dumps(req.context['doc'])
class JSONTranslator(object):
def process_request(self, req, resp):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
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):
if 'result' not in req.context:
return
resp.body = json.dumps(req.context['result'])
def max_body(limit):
def hook(req, resp, resource, params):
length = req.content_length
if length is not None and length > limit:
msg = ('The size of the request is too large. The body must not '
'exceed ' + str(limit) + ' bytes in length.')
raise falcon.HTTPRequestEntityTooLarge(
'Request body is too large', msg)
return hook
class ThingsResource:
@@ -109,7 +141,6 @@ class ThingsResource:
self.db = db
self.logger = logging.getLogger('thingsapp.' + __name__)
@falcon.after(serialize)
def on_get(self, req, resp, user_id):
marker = req.get_param('marker') or ''
limit = req.get_param_as_int('limit') or 50
@@ -132,13 +163,20 @@ class ThingsResource:
# create a custom class that inherits from falcon.Request. This
# class could, for example, have an additional 'doc' property
# that would serialize to JSON under the covers.
req.context['doc'] = result
req.context['result'] = result
resp.set_header('X-Powered-By', 'Small Furry Creatures')
resp.status = falcon.HTTP_200
@falcon.before(deserialize)
def on_post(self, req, resp, user_id, doc):
@falcon.before(max_body(64 * 1024))
def on_post(self, req, resp, user_id):
try:
doc = req.context['doc']
except KeyError:
raise falcon.HTTPBadRequest(
'Missing thing',
'A thing must be submitted in the request body.')
proper_thing = self.db.add_thing(doc)
resp.status = falcon.HTTP_201
@@ -146,7 +184,11 @@ class ThingsResource:
# Configure your WSGI server to load "things.app" (app is a WSGI callable)
app = falcon.API(before=[auth, check_media_type])
app = falcon.API(middleware=[
AuthMiddleware(),
RequireJSON(),
JSONTranslator(),
])
db = StorageEngine()
things = ThingsResource(db)
@@ -160,7 +202,7 @@ app.add_error_handler(StorageError, StorageError.handle)
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter()
app.add_sink(sink, r'/v1/[charts|inventory]')
app.add_sink(sink, r'/search/(?P<engine>ddg|y)\Z')
# Useful for debugging problems in your API; works with pdb.set_trace()
if __name__ == '__main__':