import json import pathlib from typing import Any, cast import click import yaml from click.exceptions import Exit from fastapi import FastAPI from pyxdi import PyxDI from campaigns.cli.decorators import pass_di from campaigns.config import AppSettings cli = click.Group(short_help="Perform API commands.", no_args_is_help=True) @cli.command("gen-spec") @pass_di def gen_spec(di: PyxDI) -> None: """ Generate OpenAPI specification. """ from campaigns.api.app import get_application app = get_application(di=di) settings = di.get_instance(AppSettings) dump_schema(app, filepath=settings.openapi_filepath) click.secho( ( f"OpenAPI specification was successfully " f"generated to `{settings.openapi_filepath}`", ), fg="green", ) @cli.command("check-spec") @pass_di def check_spec(di: PyxDI) -> None: """ Check that service has the latest OpenAPI specification. """ from campaigns.api.app import get_application app = get_application(di=di) settings = di.get_instance(AppSettings) if not check_schema(app, filepath=settings.openapi_filepath): click.secho( "The OpenAPI specification for the service does not match the actual one. " "Please, generate new one with `python manage.py api gen-spec`.", fg="red", ) raise Exit(1) click.secho("You have the latest OpenAPI specification generated.", fg="green") def dump_schema(api: FastAPI, *, filepath: pathlib.Path) -> None: with filepath.open("w+") as fp: yaml.dump(json.loads(json.dumps(api.openapi())), stream=fp, sort_keys=False) def check_schema(api: FastAPI, *, filepath: pathlib.Path) -> bool: with filepath.open() as fp: return cast(dict[str, Any], yaml.safe_load(fp)) == api.openapi()