Skip to content

serve

faststream.cli.docs.app.serve #

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"
    ),
    reload: bool = typer.Option(
        False,
        "--reload",
        is_flag=True,
        help="Restart documentation at directory files changes",
    ),
) -> None

Serve project AsyncAPI schema

Source code in faststream/cli/docs/app.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@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",
    ),
    reload: bool = typer.Option(
        False,
        "--reload",
        is_flag=True,
        help="Restart documentation at directory files changes",
    ),
) -> None:
    """Serve project AsyncAPI schema"""

    if ":" in app:
        module, _ = import_from_string(app)

        module_parent = module.parent
        extra_extensions: Sequence[str] = ()

    else:
        module_parent = Path.cwd()
        schema_filepath = module_parent / app
        extra_extensions = (schema_filepath.suffix,)

    if reload is True:
        try:
            from faststream.cli.supervisors.watchfiles import WatchReloader

        except ImportError:
            warnings.warn(INSTALL_WATCHFILES, category=ImportWarning, stacklevel=1)
            _parse_and_serve(app, host, port)

        else:
            WatchReloader(
                target=_parse_and_serve,
                args=(app, host, port),
                reload_dirs=(str(module_parent),),
                extra_extensions=extra_extensions,
            ).run()

    else:
        _parse_and_serve(app, host, port)