Examples¶
flask¶
from dataclasses import asdict, dataclass
from time import perf_counter
from flask import Flask, request
from defspec import OpenAPI, RenderTemplate
@dataclass
class JSONRequest:
title: str
timeout: float
@dataclass
class Query:
limit: int
offset: int
@dataclass
class JSONResponse:
elapsed: float
queries: list[Query]
app = Flask(__name__)
@app.route("/fake", methods=["POST"])
def fake():
start = perf_counter()
req: JSONRequest = JSONRequest(**request.json)
print(req)
query: Query = Query(**request.args)
response = JSONResponse(elapsed=perf_counter() - start, queries=[query])
return asdict(response)
@app.route("/", methods=["GET"])
def health_check():
return "OK"
@app.route("/openapi/spec.json", methods=["GET"])
def spec():
openapi = OpenAPI()
openapi.register_route("/", "get", summary="health check")
openapi.register_route(
"/fake",
"post",
summary="fake endpoint",
request_type=JSONRequest,
response_type=JSONResponse,
query_type=Query,
)
return openapi.to_dict()
@app.route("/openapi/redoc", methods=["GET"])
def redoc():
return RenderTemplate.REDOC.value.format(spec_url="/openapi/spec.json")
@app.route("/openapi/swagger", methods=["GET"])
def swagger():
return RenderTemplate.SWAGGER.value.format(spec_url="/openapi/spec.json")
@app.route("/openapi/scalar", methods=["GET"])
def scalar():
return RenderTemplate.SCALAR.value.format(spec_url="/openapi/spec.json")
if __name__ == "__main__":
app.run(debug=True, port=8000)
falcon¶
from time import perf_counter
from wsgiref.simple_server import make_server
import falcon
import msgspec
from falcon import App, Request, Response
from defspec import OpenAPI, RenderTemplate
class JSONRequest(msgspec.Struct, frozen=True):
title: str
timeout: float
class Query(msgspec.Struct, frozen=True):
limit: int
offset: int
class JSONResponse(msgspec.Struct, frozen=True):
elapsed: float
queries: list[Query]
class FakeResource:
def on_post(self, req: Request, resp: Response):
start = perf_counter()
request: JSONRequest = msgspec.json.decode(
req.bounded_stream.read(), type=JSONRequest, strict=False
)
print(request)
query: Query = msgspec.convert(req.params, Query, strict=False)
response = JSONResponse(elapsed=perf_counter() - start, queries=[query])
resp.data = msgspec.json.encode(response)
class HealthCheck:
def on_get(self, req: Request, resp: Response):
resp.text = "OK"
resp.content_type = falcon.MEDIA_TEXT
class OpenAPIResource:
def __init__(self) -> None:
self.openapi = OpenAPI()
self.openapi.register_route("/", "get", summary="health check")
self.openapi.register_route(
"/fake",
"post",
summary="fake endpoint",
request_type=JSONRequest,
response_type=JSONResponse,
query_type=Query,
)
self.spec = self.openapi.to_json()
def on_get(self, req: Request, resp: Response):
resp.content_type = falcon.MEDIA_JSON
resp.data = self.spec
class OpenAPIRender:
def __init__(self, spec_url: str, template: RenderTemplate) -> None:
self.template = template.value.format(spec_url=spec_url)
def on_get(self, req: Request, resp: Response):
resp.content_type = falcon.MEDIA_HTML
resp.text = self.template
if __name__ == "__main__":
app = App()
app.add_route("/", HealthCheck())
app.add_route("/fake", FakeResource())
app.add_route("/openapi/spec.json", OpenAPIResource())
app.add_route(
"/openapi/redoc", OpenAPIRender("/openapi/spec.json", RenderTemplate.REDOC)
)
app.add_route(
"/openapi/swagger", OpenAPIRender("/openapi/spec.json", RenderTemplate.SWAGGER)
)
app.add_route(
"/openapi/scalar", OpenAPIRender("/openapi/spec.json", RenderTemplate.SCALAR)
)
with make_server("", 8000, app) as server:
print("Serving on port 8000...")
server.serve_forever()
offline usage¶
from time import perf_counter
from wsgiref.simple_server import make_server
import falcon
import msgspec
from falcon import App, Request, Response
from offapi import OpenAPITemplate
from defspec import OpenAPI
class JSONRequest(msgspec.Struct, frozen=True):
title: str
timeout: float
class Query(msgspec.Struct, frozen=True):
limit: int
offset: int
class JSONResponse(msgspec.Struct, frozen=True):
elapsed: float
queries: list[Query]
class FakeResource:
def on_post(self, req: Request, resp: Response):
start = perf_counter()
request: JSONRequest = msgspec.json.decode(
req.bounded_stream.read(), type=JSONRequest, strict=False
)
print(request)
query: Query = msgspec.convert(req.params, Query, strict=False)
response = JSONResponse(elapsed=perf_counter() - start, queries=[query])
resp.data = msgspec.json.encode(response)
class HealthCheck:
def on_get(self, req: Request, resp: Response):
resp.text = "OK"
resp.content_type = falcon.MEDIA_TEXT
class OpenAPIResource:
def __init__(self) -> None:
self.openapi = OpenAPI()
self.openapi.register_route("/", "get", summary="health check")
self.openapi.register_route(
"/fake",
"post",
summary="fake endpoint",
request_type=JSONRequest,
response_type=JSONResponse,
query_type=Query,
)
self.spec = self.openapi.to_json()
def on_get(self, req: Request, resp: Response):
resp.content_type = falcon.MEDIA_JSON
resp.data = self.spec
class OpenAPIRender:
def __init__(self, spec_url: str, template: OpenAPITemplate) -> None:
self.template = template.value.format(spec_url=spec_url)
def on_get(self, req: Request, resp: Response):
resp.content_type = falcon.MEDIA_HTML
resp.text = self.template
if __name__ == "__main__":
app = App()
app.add_route("/", HealthCheck())
app.add_route("/fake", FakeResource())
app.add_route("/openapi/spec.json", OpenAPIResource())
app.add_route(
"/openapi/redoc", OpenAPIRender("/openapi/spec.json", OpenAPITemplate.REDOC)
)
app.add_route(
"/openapi/swagger", OpenAPIRender("/openapi/spec.json", OpenAPITemplate.SWAGGER)
)
app.add_route(
"/openapi/scalar", OpenAPIRender("/openapi/spec.json", OpenAPITemplate.SCALAR)
)
with make_server("", 8000, app) as server:
print("Serving on port 8000...")
server.serve_forever()