Add schema command

Author: mlasevich

README.md

@@ -102,6 +102,12 @@ It will listen on the IP address `0.0.0.0`, which means all the available IP add
 
 In most cases you would (and should) have a "termination proxy" handling HTTPS for you on top, this will depend on how you deploy your application, your provider might do this for you, or you might need to set it up yourself. You can learn more about it in the <a href="https://fastapi.tiangolo.com/deployment/" class="external-link" target="_blank">FastAPI Deployment documentation</a>.
 
+## `fastapi schema`
+
+When you run `fastapi schema`, it will generate a swagger/openapi document.
+
+This document will be output to stderr by default, however `--output <filename>` option can be used to write output into file. You can control the format of the JSON file by specifying indent level with `--indent #`. If set to 0, JSON will be in the minimal/compress form. Default is 2 spaces. 
+
 ## License
 
 This project is licensed under the terms of the MIT license.

src/fastapi_cli/cli.py

@@ -1,3 +1,5 @@
+import json
+import sys
 from logging import getLogger
 from pathlib import Path
 from typing import Any, Union
@@ -9,6 +11,7 @@
 from typing_extensions import Annotated
 
 from fastapi_cli.discover import get_import_string
+from fastapi_cli.discover import get_app
 from fastapi_cli.exceptions import FastAPICLIException
 
 from . import __version__
@@ -272,6 +275,43 @@ def run(
         proxy_headers=proxy_headers,
     )
 
+@app.command()
+def schema(
+        path: Annotated[
+            Union[Path, None],
+            typer.Argument(
+                help="A path to a Python file or package directory (with [blue]__init__.py[/blue] files) containing a [bold]FastAPI[/bold] app. If not provided, a default set of paths will be tried."
+            ),
+        ] = None,
+        *,
+        app: Annotated[
+            Union[str, None],
+            typer.Option(
+                help="The name of the variable that contains the [bold]FastAPI[/bold] app in the imported module or package. If not provided, it is detected automatically."
+            ),
+        ] = None,
+        output: Annotated[
+            Union[str, None],
+            typer.Option(
+                help="The filename to write schema to. If not provided, write to stderr."
+            ),
+        ] = None,
+        indent: Annotated[
+            int,
+            typer.Option(
+                help="JSON format indent. If 0, disable pretty printing"
+            ),
+        ] = 2,
+        ) -> Any:
+    """ Generate schema """
+    app = get_app(path=path, app_name=app)
+    schema = app.openapi()
+
+    stream = open(output, "w") if output else sys.stderr
+    json.dump(schema, stream, indent=indent if indent > 0 else None)
+    if output:
+        stream.close()
+    return 0
 
 def main() -> None:
     app()

src/fastapi_cli/discover.py

@@ -1,5 +1,6 @@
 import importlib
 import sys
+from contextlib import contextmanager
 from dataclasses import dataclass
 from logging import getLogger
 from pathlib import Path
@@ -46,6 +47,18 @@ class ModuleData:
     module_import_str: str
     extra_sys_path: Path
 
+    @contextmanager
+    def sys_path(self):
+        """ Context manager to temporarily alter sys.path"""
+        extra_sys_path = str(self.extra_sys_path) if self.extra_sys_path else ""
+        if extra_sys_path:
+            logger.warning("Adding %s to sys.path...", extra_sys_path)
+            sys.path.insert(0, extra_sys_path)
+            yield
+        if extra_sys_path and sys.path and sys.path[0] == extra_sys_path:
+            logger.warning("Removing %s from sys.path...", extra_sys_path)
+            sys.path.pop(0)
+
 
 def get_module_data_from_path(path: Path) -> ModuleData:
     logger.info(
@@ -165,3 +178,54 @@ def get_import_string(
     import_string = f"{mod_data.module_import_str}:{use_app_name}"
     logger.info(f"Using import string [b green]{import_string}[/b green]")
     return import_string
+
+def get_app(
+        *, path: Union[Path, None] = None, app_name: Union[str, None] = None
+) -> FastAPI:
+    if not path:
+        path = get_default_path()
+    logger.debug(f"Using path [blue]{path}[/blue]")
+    logger.debug(f"Resolved absolute path {path.resolve()}")
+    if not path.exists():
+        raise FastAPICLIException(f"Path does not exist {path}")
+    mod_data = get_module_data_from_path(path)
+    try:
+        with mod_data.sys_path():
+            mod = importlib.import_module(mod_data.module_import_str)
+    except (ImportError, ValueError) as e:
+        logger.error(f"Import error: {e}")
+        logger.warning(
+            "Ensure all the package directories have an [blue]__init__.py["
+            "/blue] file"
+        )
+        raise
+    if not FastAPI:  # type: ignore[truthy-function]
+        raise FastAPICLIException(
+            "Could not import FastAPI, try running 'pip install fastapi'"
+        ) from None
+    object_names = dir(mod)
+    object_names_set = set(object_names)
+    if app_name:
+        if app_name not in object_names_set:
+            raise FastAPICLIException(
+                f"Could not find app name {app_name} in "
+                f"{mod_data.module_import_str}"
+            )
+        app = getattr(mod, app_name)
+        if not isinstance(app, FastAPI):
+            raise FastAPICLIException(
+                f"The app name {app_name} in {mod_data.module_import_str} "
+                f"doesn't seem to be a FastAPI app"
+            )
+        return app
+    for preferred_name in ["app", "api"]:
+        if preferred_name in object_names_set:
+            obj = getattr(mod, preferred_name)
+            if isinstance(obj, FastAPI):
+                return obj
+    for name in object_names:
+        obj = getattr(mod, name)
+        if isinstance(obj, FastAPI):
+            return obj
+    raise FastAPICLIException(
+        "Could not find FastAPI app in module, try using --app")

tests/test_cli.py

@@ -179,6 +179,15 @@ def test_dev_help() -> None:
     assert "The name of the variable that contains the FastAPI app" in result.output
     assert "Use multiple worker processes." not in result.output
 
+def test_schema() -> None:
+    with changing_dir(assets_path):
+        with open('openapi.json', 'r') as stream:
+            expected = stream.read()
+        assert expected != "" , "Failed to read expected result"
+        result = runner.invoke(app, ["schema", "single_file_app.py"])
+        assert result.exit_code == 0, result.output
+        assert expected in result.output,  result.output
+
 
 def test_run_help() -> None:
     result = runner.invoke(app, ["run", "--help"])