fastapi openapi.yaml in addition to openapi.json + hook openapi.yaml into redoc page

_fastapi-oas-yaml.md

Add openapi.yaml as alternative OAS schema format in fastapi (https://fastapi.tiangolo.com/), as /hello-world-service.yaml in this case.

Moreover, put this endpoint into the redoc page instead of the link to the .json OAS schema file.

See also this fastapi feature request: tiangolo/fastapi#1140

app.yaml

fastapi:
  app:
    services:
      - service:
          name: hello-world-service
          path: /hello-world-service

    docs:
      openapi_url: /hello-world-service-oas.json

main.py

from fastapi import FastAPI
from fastapi.openapi.docs import get_redoc_html
from fastapi.responses import Response
from fastapi.responses import PlainTextResponse
import functools
import io
import os
import os.path
from pydantic import BaseModel
import re
from starlette.requests import Request
from starlette.responses import HTMLResponse
import yaml


def read_config(app_config=None):
    if app_config is None:
        app_dir = os.path.dirname(__file__)
        app_config = os.path.join(app_dir, 'app.yaml')
    with open(app_config, 'r') as cfgfile:
        config = yaml.safe_load(cfgfile)
    return config


class DPath:
    """Access dictionary paths as mydict.key.subkey "dotted attribute paths".
    """

    _sentinel = object()
    _split = re.compile('[.\[\]]').split

    @classmethod
    def _int_or_string(cls, s):
        try:
            return int(s)
        except ValueError:
            return s

    @classmethod
    def split(cls, path):
        return [cls._int_or_string(p) for p in cls._split(path) if p]

    def __init__(self, path):
        self.path = self.split(path)

    def get(self, dct, default=_sentinel):
        try:
            val = dct
            for pathseg in self.path:
               val = val[pathseg]
            return val
        except KeyError:
            if default is self._sentinel:
                raise
            return default


config = read_config()
service = DPath('fastapi.app.services[0].service').get(config)
service_name = service['name']
service_path = service['path'].rstrip('/')


def p(path):
    # a helper that prefixes service_path to the service endpoint routes
    return '/'.join([service_path, path.lstrip('/')])

   
openapi_url_json = DPath(
    'fastapi.app.docs.openapi_url').get(config, '/openapi.json')
openapi_url_yaml = ''.join([os.path.splitext(openapi_url_json)[0], '.yaml'])
docs_url = DPath('fastapi.app.docs.docs_url').get(config, '/doc')
redoc_url = DPath('fastapi.app.docs.docs_url').get(config, '/redoc')


app = FastAPI(
    title=service_name,
    docs_url=p(docs_url),
    # supppress redoc endpoint initially, we'll add it later with yaml
    # customization
    redoc_url=None,
    openapi_url=p(openapi_url_json),
    )


# data model
class HelloResponse(BaseModel):
    message: str


class HealthResponse(BaseModel):
    # could be an enum
    status: str


# add endpoints
# additional yaml version of openapi.json
# TODO: Is the lru_cache applicable here due to async trickery? Must investigate...
@app.get(p(openapi_url_yaml), include_in_schema=False)
@functools.lru_cache()
def read_openapi_yaml() -> Response:
    openapi_json= app.openapi()
    yaml_s = io.StringIO()
    yaml.dump(openapi_json, yaml_s)
    return Response(yaml_s.getvalue(), media_type='text/yaml')


# hook openapi.yaml instead of openapi.json into redoc html page
@app.get(p(redoc_url), include_in_schema=False)
async def redoc_html(req: Request) -> HTMLResponse:
    root_path = req.scope.get("root_path", "").rstrip("/")
    openapi_url = root_path + p(openapi_url_yaml)
    return get_redoc_html(
        openapi_url=openapi_url,
        title=app.title + " - ReDoc"
        )


@app.get(p("/"), response_class=PlainTextResponse)
def read_root():
    return "The Hello world service!"


@app.get(p("/hello"), response_model=HelloResponse)
def read_hello():
    return {"message": "Hello, world!"}


@app.get(p("/hello/{name}"), response_model=HelloResponse)
def read_hello_name(name: str):
    return {"message": f"Hello, {name}!"}


@app.get(p("/health"), response_model=HealthResponse)
def read_health():
    return {"status": "UP"}