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:

  1. POST http://127.0.0.1:8000/account/1234?price=40

  2. GET http://127.0.0.1:8000/invalid-params

  3. GET http://127.0.0.1:8000/invalid-params?age=-10&color

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.

  1. POST http://127.0.0.1:8000/ to create a new customer

  2. GET http://127.0.0.1:8000/<id> to retrieve a customer

  3. GET http://127.0.0.1:8000/ to view the API documentation

  4. 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()