Skip to content

serve

faststream.cli.docs.app.serve #

serve(app=Argument(..., help='[python_module:FastStream] or [asyncapi.yaml/.json] - path to your application or documentation.'), host=Option('localhost', help='Documentation hosting address.'), port=Option(8000, help='Documentation hosting port.'), reload=Option(False, '--reload', is_flag=True, help='Restart documentation at directory files changes.'), app_dir=Option('.', '--app-dir', help='Look for APP in the specified directory, by adding this to the PYTHONPATH. Defaults to the current working directory.'), is_factory=Option(False, '--factory', is_flag=True, help='Treat APP as an application factory.'))

Serve project AsyncAPI schema.

Source code in faststream/cli/docs/app.py 
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@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.",
    ),
    app_dir: str = typer.Option(
        ".",
        "--app-dir",
        help=(
            "Look for APP in the specified directory, by adding this to the PYTHONPATH."
            " Defaults to the current working directory."
        ),
    ),
    is_factory: bool = typer.Option(
        False,
        "--factory",
        is_flag=True,
        help="Treat APP as an application factory.",
    ),
) -> None:
    """Serve project AsyncAPI schema."""
    if ":" in app:
        if app_dir:  # pragma: no branch
            sys.path.insert(0, app_dir)

        module, _ = import_from_string(app, is_factory=is_factory)

        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:
        try:
            from faststream.cli.supervisors.watchfiles import WatchReloader

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

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

    else:
        _parse_and_serve(app, host, port, is_factory)