forked from gothinkster/realworld-starter-kit
-
Notifications
You must be signed in to change notification settings - Fork 21
/
openapi.py
106 lines (78 loc) · 3.56 KB
/
openapi.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
"""Integration with pyramid_openapi3."""
from datetime import datetime
from pyramid.config import Configurator
from pyramid.httpexceptions import exception_response
from pyramid.httpexceptions import HTTPNotFound
from pyramid.httpexceptions import HTTPUnauthorized
from pyramid.httpexceptions import HTTPUnprocessableEntity
from pyramid.renderers import JSON
from pyramid.request import Request
from pyramid.view import exception_view_config
from pyramid.view import forbidden_view_config
from pyramid.view import notfound_view_config
import os
import structlog
import typing as t
logger = structlog.get_logger(__name__)
def includeme(config: Configurator) -> None:
"""Configure support for serving and handling OpenAPI requests."""
config.include("pyramid_openapi3")
config.pyramid_openapi3_spec(
os.path.join(os.path.dirname(__file__), "openapi.yaml"), route="/openapi.yaml"
)
config.pyramid_openapi3_add_explorer(route="/api")
config.add_renderer("json", json_renderer())
def json_renderer() -> JSON:
"""Configure a JSON renderer that supports rendering datetimes."""
renderer = JSON()
renderer.add_adapter(datetime, datetime_adapter)
return renderer
def datetime_adapter(obj: datetime, request: Request) -> str:
"""OpenAPI spec defines date-time notation as RFC 3339, section 5.6. # noqa
For example: 2017-07-21T17:32:28.001Z
The `timespec="milliseconds"` is required because the frontend expects
the format to be exactly `HH:MM:SS.sss` and not `HH:MM:SS` or
`HH:MM:SS.ssssss` which Python would decide automatically.
"""
return obj.isoformat(timespec="milliseconds") + "Z"
def object_or_404(obj: t.Any) -> t.Any:
"""Use this to stop view code execution if object is not found.
If obj is None, return response with 404 status code.
Otherwise, continue with view code.
Example usage:
user = object_or_404(User.by_username("foo", db=request.db))
... # code here knows that user exists and does not need to do any checks
"""
if obj is None:
raise HTTPNotFound
else:
return obj
@exception_view_config(Exception)
def unknown_error(exc: Exception, request: Request) -> HTTPUnprocessableEntity:
"""Catch any uncaught errors and respond with a nice JSON error.
Without this, Exceptions that get to WSGI level return HTML that says
Internal Server Error.
Note that, in an ideal world, this view would return 500 instead of 422,
as an unexpected error means something is wrong with the server, hence
500 Internal Server Error makes more sense. Alas, this is a RealWorld.io
example implementation, and the specs say it should be 422:
https://github.com/gothinkster/realworld/blob/master/api/swagger.json
"""
logger.exception("Uncaught error", exc_info=exc)
return exception_response(
422, json_body={"errors": {"body": ["Internal Server Error"]}}
)
@forbidden_view_config()
def unauthorized(request: Request) -> HTTPUnauthorized:
"""Catch permission errors and respond with a nice JSON error.
Without this, permission errors that get to WSGI level return HTML that says
403 Forbidden.
"""
return exception_response(401, json_body={"errors": {"body": ["Unauthorized"]}})
@notfound_view_config()
def notfound(request: Request) -> HTTPNotFound:
"""Catch url traversal and db lookup errors and respond with a nice JSON error.
Without this, not-found errors that get to WSGI level return HTML that says
404 Not Found.
"""
return exception_response(404, json_body={"errors": {"body": ["Not Found"]}})