serve(
app: str = typer.Argument(
...,
help="[python_module:FastStream] or [asyncapi.yaml/.json] - path to your application or documentation",
),
host: str = typer.Option(
"localhost", help="documentation hosting address"
),
port: int = typer.Option(
8000, help="documentation hosting port"
),
) -> None
Serve project AsyncAPI schema
Source code in faststream/cli/docs/app.py
| @docs_app.command(name="serve")
def serve(
app: str = typer.Argument(
...,
help="[python_module:FastStream] or [asyncapi.yaml/.json] - path to your application or documentation",
),
host: str = typer.Option(
"localhost",
help="documentation hosting address",
),
port: int = typer.Option(
8000,
help="documentation hosting port",
),
) -> None:
"""Serve project AsyncAPI schema"""
if ":" in app:
_, app_obj = import_from_string(app)
raw_schema = get_app_schema(app_obj)
else:
schema_filepath = Path.cwd() / app
if schema_filepath.suffix == ".json":
data = schema_filepath.read_text()
elif schema_filepath.suffix == ".yaml" or schema_filepath.suffix == ".yml":
try:
import yaml
except ImportError as e: # pragma: no cover
typer.echo(INSTALL_YAML, err=True)
raise typer.Exit(1) from e
with schema_filepath.open("r") as f:
schema = yaml.safe_load(f)
data = json.dumps(schema)
else:
raise ValueError(
f"Unknown extension given - {app}; Please provide app in format [python_module:FastStream] or [asyncapi.yaml/.json] - path to your application or documentation"
)
raw_schema = model_parse(Schema, data)
serve_app(
schema=raw_schema,
host=host,
port=port,
)
|