Examples¶
You can run any of the examples in this section by preparing a Python environment:
$ python3 -mvenv env
$ env/bin/pip install -q '.[examples]'
$ env/bin/python examples/httpbin.py
DEBUG asyncio: Using selector: KqueueSelector
INFO root: listening on http://127.0.0.1:8000
INFO root: try me at http://127.0.0.1:8000/?status=500
The examples run the application in debug mode so changing the source code results in the application reloading immediately. Feel free to play with the examples.
httpbin.org¶
This example sends a request to httpbin.org that generates a HTTP failure.
The failure is caught as a tornado.web.HTTPError
instance and
translated into a problemdetails.Problem
instance. You can run
this example with python examples/httpbin.py
and send requests to
http://localhost:8000/?status=500
.
async def get(self):
logger = logging.getLogger('HttpBinHandler')
url = 'http://httpbin.org/status/{0}'.format(
self.get_query_argument('status', '500'))
logger.info('retrieving %s', url)
client = httpclient.AsyncHTTPClient()
try:
response = await client.fetch(url)
self.add_header('Content-Type', 'application/json')
self.write(response.body)
except httpclient.HTTPError as error:
raise problemdetails.Problem(
status_code=error.code,
httpbin_headers=dict(error.response.headers.items()))
RFC-7807 examples¶
This example includes two endpoints that return the examples from RFC 7807:
The first example returns the 403 credit error from section 3. This example exemplifies raising an exception where the failure is detected and finishing the request processing by not catching the exception. The raised exception is fairly complex but constructed from literals.
class AccountHandler(problemdetails.ErrorWriter, web.RequestHandler):
def post(self, account):
price = float(self.get_query_argument('price'))
self.verify_funds(account, price)
self.buy_thing()
def verify_funds(self, account, price):
balance = self.get_balance(account)
if price > balance:
raise problemdetails.Problem(
status_code=403,
title='You do not have enough credit.',
detail=(f'Your current balance is {balance}, but that '
f'costs {price}'),
instance=self.reverse_url('account-handler', account),
balance=balance,
accounts=self.lookup_accounts(account),
type='https://example.com/probs/out-of-credit',
)
def get_balance(self, account):
return 30.0 # TODO look up balance
def lookup_accounts(self, primary_account):
related = [12345, 67890] # TODO lookup related accounts
return [self.reverse_url('account-handler', aid) for aid in related]
def buy_thing(self):
pass
The second example returns the more complex 400 invalid parameter error from section 3. The errors are accumulated into a list and raised from the top-level after the parameters are extracted. This is a different approach where all of the error handling is explicit and at the top-level.
class ValidationError(problemdetails.ErrorWriter, web.RequestHandler):
def initialize(self):
super(ValidationError, self).initialize()
self.errors = []
def get(self):
age = self._extract_age()
color = self._extract_color()
if self.errors:
raise problemdetails.Problem(**{
'status_code': 400,
'type': 'https://example.net/validation-error',
'invalid-params': self.errors,
})
self.write({'age': age, 'color': color})
def _record_error(self, field, error):
self.errors.append({'name': field, 'reason': error})
def _extract_age(self):
try:
age = self.get_query_argument('age')
except web.MissingArgumentError:
self._record_error('age', 'is required')
else:
try:
age = int(age)
if age > 0:
return age
self._record_error('age', 'must be a positive integer')
except ValueError:
self._record_error('age', 'must be a integer')
return None
def _extract_color(self):
try:
color = self.get_query_argument('color')
except web.MissingArgumentError:
self._record_error('color', 'is required')
else:
if color in ('blue', 'green', 'red'):
return color
self._record_error('color', "must be 'green', 'red' or 'blue'")
return None
JSON Schema and Open API¶
This example implements a very simple create & retrieve interface that could be the beginning of a customer data service. It includes a complete Open API 3 schema, a documentation endpoint that renders the API, and JSON schema validation of incoming requests. Since it is an example, everything is in one file including the Open API specification, the data store implementation, and all of the handlers.
POST http://127.0.0.1:8000/ to create a new customer
GET http://127.0.0.1:8000/<id> to retrieve a customer
GET http://127.0.0.1:8000/ to view the API documentation
GET http://127.0.0.1:8000/openapi.json for the machine-readable specification
The Open API specification is embedded as a long literal YAML string. It
is deserialized and stored in the application settings so handlers refer
to it using self.settings['openapi']
. The creation handler uses the
JSON schema embedded in the API specification to validate the request
entity and transforms JSON schema validation errors into a readable error
document.
The customer creation handler is the most interesting so let’s start there first.
1 def post(self):
2 content_type, params = cgi.parse_header(
3 self.request.headers.get('Content-Type', 'application/json'))
4 if content_type != 'application/json':
5 raise problemdetails.Problem(
6 status_code=415,
7 title='Content type is not understood',
8 detail='Cannot decode {0}, try application/json'.format(
9 content_type),
10 )
11
12 try:
13 body = json.loads(self.request.body.decode('utf-8'))
14 except (TypeError, ValueError, UnicodeDecodeError) as error:
15 raise problemdetails.Problem(
16 status_code=400,
17 title='Failed to decode request',
18 detail=str(error),
19 failure=error,
20 )
21
22 openapi = self.settings['openapi']
23 validator = jsonschema.Draft7Validator(
24 schema=openapi['components']['schemas']['CustomerDetails'])
25 all_errors = list(validator.iter_errors(body))
26 most_important = jsonschema.exceptions.best_match(all_errors)
27 if most_important is not None:
28 raise problemdetails.Problem(
29 status_code=422,
30 type='/errors#jsonschema-failure',
31 title='Failed to process request',
32 detail=most_important.message,
33 failure=all_errors,
34 )
35
36 record_id = self.settings['database'].insert(body)
37 self.redirect(self.reverse_url('retrieve', record_id), status=303)
Lines 2-10 verify that the incoming request is a JSON entity. If not, it generates a 415 Unsupported Media Type response. Lines 12-20 decode the incoming request body and generates a 400 Bad Request if it fails to JSON decode the body.
Lines 22-34 are the interesting ones. The Open API specification is
available as self.settings['openapi']
and the incoming message schema
is in components/schemas/CustomerDetails. The message schema is used to
validate the incoming request as described in the python-jsonschema
documentation. jsonschema.exceptions.best_match()
function will
return the most important error from all_errors
if the request body
failed to validate. In the case, a 422 Unprocessable Request is returned
with a lot of detail embedded in the problem document.
If it receives a well-formed JSON document that is lacking information then it responds with the expected mixture of human-readable and machine-processable information.
POST / HTTP/1.1
Content-Type: application/json; charset=UTF-8
Host: 127.0.0.1:8000
{"name":"Dave Shawley","address":{"address":"somewhere"}}
HTTP/1.1 422 Unprocessable Entity
Content-Length: 426
Content-Type: application/problem+json
Date: Mon, 08 Apr 2019 13:36:15 GMT
Server: TornadoServer/6.0.2
{
"detail": "'email' is a required property",
"failure": [
{
"absolute_path": [
"address"
],
"absolute_schema_path": [
"properties",
"address",
"required"
],
"context": [],
"message": "'country' is a required property"
},
{
"absolute_path": [],
"absolute_schema_path": [
"required"
],
"context": [],
"message": "'email' is a required property"
}
],
"status": 422,
"title": "Failed to process request",
"type": "/errors#jsonschema-failure"
}
The last piece that requires some explanation is the slight customization
to the JSON encoder. A custom “default object handler” is installed into
the error writer’s JSON encoder. The object handler is shown below. It
knows how to translate ValidationError
instances into readable documents.
def jsonify(obj):
"""Transform `obj` into something that the JSON encoder can handle."""
if isinstance(obj, uuid.UUID):
return str(obj)
if isinstance(obj, jsonschema.ValidationError):
return {
'absolute_path': list(obj.absolute_path),
'absolute_schema_path': list(obj.absolute_schema_path),
'context': obj.context,
'message': obj.message,
}
if isinstance(obj, Exception):
return {
'exception': obj.__class__.__name__,
'message': str(obj),
'arguments': obj.args,
}
raise TypeError('{0} is not JSON serializable'.format(
obj.__class__.__name__))
The JSON document generation is customized by setting the default
attribute on the problemdetails.ErrorWriter.json_encoder
instance
when the application is created.
async def main():
app = web.Application([
web.url(r'/index.html', DocumentationHandler, name='docs'),
web.url(r'/openapi.json', OpenAPIHandler, name='openapi'),
web.url(r'/', CreateHandler, name='create'),
web.url(r'/(?P<record_id>.+)', FetchHandler, name='retrieve'),
])
app.settings['database'] = CustomerDatabase()
app.settings['debug'] = True
app.settings['json-encoder'] = json.JSONEncoder(default=jsonify)
app.settings['openapi'] = yaml.safe_load(OPENAPI_SCHEMA)
# Update the error writer's JSON encoder so that it knows
# how to handle validation errors
problemdetails.ErrorWriter.json_encoder.default = jsonify
port = int(os.environ.get('PORT', '8000'))
app.listen(address='127.0.0.1', port=port)
event = asyncio.Event()
loop = asyncio.get_event_loop()
loop.add_signal_handler(signal.SIGINT, event.set)
loop.add_signal_handler(signal.SIGTERM, event.set)
await event.wait()