Coverage for fastagency/cli/cli.py: 73%
42 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-19 12:16 +0000
1from logging import getLogger 1afghbicde
2from pathlib import Path 1afghbicde
3from typing import Annotated, Any, Optional 1afghbicde
5import typer 1afghbicde
7from .. import __version__ 1afghbicde
8from ..exceptions import FastAgencyCLIError, FastAgencyCLIPythonVersionError 1afghbicde
9from .discover import get_import_string 1afghbicde
10from .docker_cli import docker_app 1afghbicde
11from .logging import setup_logging 1afghbicde
13app = typer.Typer(rich_markup_mode="rich") 1afghbicde
14app.add_typer( 1afghbicde
15 docker_app,
16 name="docker",
17 help="[bold]Docker[/bold] commands for [bold]FastAgency[/bold]",
18)
20setup_logging() 1afghbicde
21logger = getLogger(__name__) 1afghbicde
24def version_callback(value: bool) -> None: 1afghbicde
25 if value: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true1abcde
26 typer.echo(f"{__version__}")
27 raise typer.Exit()
30@app.callback() 1afghbicde
31def callback( 1afghbicde
32 version: Annotated[
33 Optional[bool],
34 typer.Option(
35 "--version", help="Show the version and exit.", callback=version_callback
36 ),
37 ] = None,
38) -> None:
39 """FastAgency CLI - The [bold]fastapi[/bold] command line app. 😎
41 Manage your [bold]FastAgency[/bold] projects, run your FastAgency apps, and more.
43 Read more in the docs: [link]https://fastagency.ai/latest/[/link].
44 """ # noqa: D415
47def _run_app( 1afghbicde
48 *,
49 path: Optional[Path],
50 app: Optional[str],
51 workflow: Optional[str],
52 params: dict[str, Any],
53 dev_mode: bool = False,
54 single_run: bool = False,
55) -> None:
56 try: 1ab
57 import_string, fa_app = get_import_string(path=path, app_name=app) 1ab
59 with fa_app.create(import_string=import_string): 1b
60 fa_app.start( 1b
61 import_string=import_string,
62 name=workflow,
63 params=params,
64 single_run=single_run,
65 )
66 except FastAgencyCLIPythonVersionError as e: 1a
67 msg = e.args[0] 1a
68 typer.echo(msg, err=True) 1a
69 raise typer.Exit(code=1) # noqa: B904 1a
70 except FastAgencyCLIError as e:
71 msg = e.args[0]
72 typer.echo(msg, err=True)
73 raise typer.Exit(code=1) from None
76def _get_help_messages(dev_mode: bool = False) -> dict[str, str]: 1afghbicde
77 help = f"""Run a [bold]FastAgency[/bold] app in [yellow]{"development" if dev_mode else "production"}[/yellow] mode. 🚀 1afghbicde
79{"This is similar to the [bold]fastagency run[/bold] command but with [bold]reload[/bold] enabled and listening on the [blue]127.0.0.1[/blue] address." if dev_mode else "This is similar to the [bold]fastagency dev[/bold] command, but optimized for production environments."}
81It automatically detects the Python module or package that needs to be imported based on the file or directory path passed.
83If no path is passed, it tries with:
85- [blue]main.py[/blue]
86- [blue]app.py[/blue]
87- [blue]api.py[/blue]
88- [blue]app/main.py[/blue]
89- [blue]app/app.py[/blue]
90- [blue]app/api.py[/blue]
92It also detects the directory that needs to be added to the [bold]PYTHONPATH[/bold] to make the app importable and adds it.
94It detects the [bold]FastAgency[/bold] app object to use. By default it looks in the module or package for an object named:
96- [blue]app[/blue]
97- [blue]api[/blue]
99Otherwise, it uses the first [bold]FastAgency[/bold] app found in the imported module or package.
100"""
101 short_help = f"Run a [bold]FastAgency[/bold] app in [yellow]{'development' if dev_mode else 'production'}[/yellow] mode." 1afghbicde
102 return {"help": help, "short_help": short_help} 1afghbicde
105@app.command( 1afghbicde
106 context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
107 **_get_help_messages(False), # type: ignore[arg-type]
108)
109def run( 1afghbicde
110 path: Annotated[
111 Optional[Path],
112 typer.Argument(
113 help="A path to a Python file or package directory (with [blue]__init__.py[/blue] files) containing a [bold]FastAgency[/bold] app. If not provided, a default set of paths will be tried."
114 ),
115 ] = None,
116 *,
117 app: Annotated[
118 Optional[str],
119 typer.Option(
120 help="The name of the variable that contains the [bold][/bold] app in the imported module or package. If not provided, it is detected automatically."
121 ),
122 ] = None,
123 workflow: Annotated[
124 Optional[str],
125 typer.Option(
126 "--workflow",
127 "-w",
128 help="The name of the workflow to run. If not provided, the default workflow will be run.",
129 ),
130 ] = None,
131 single_run: Annotated[
132 bool,
133 typer.Option(
134 "--single-run", help="If set, only a single workflow will be executed."
135 ),
136 ] = False,
137 ctx: typer.Context,
138) -> None:
139 if len(ctx.args) > 0: 1ab
140 raise NotImplementedError("Extra arguments are not supported in this command.")
141 else:
142 params: dict[str, Any] = {} 1ab
144 dev_mode = False 1ab
145 _run_app( 1ab
146 path=path,
147 app=app,
148 workflow=workflow,
149 params=params,
150 dev_mode=dev_mode,
151 single_run=single_run,
152 )
155@app.command( 1afghbicde
156 context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
157 **_get_help_messages(True), # type: ignore[arg-type]
158)
159def dev( 1afghbicde
160 path: Annotated[
161 Optional[Path],
162 typer.Argument(
163 help="A path to a Python file or package directory (with [blue]__init__.py[/blue] files) containing a [bold]FastAgency[/bold] app. If not provided, a default set of paths will be tried."
164 ),
165 ] = None,
166 *,
167 app: Annotated[
168 Optional[str],
169 typer.Option(
170 help="The name of the variable that contains the [bold][/bold] app in the imported module or package. If not provided, it is detected automatically."
171 ),
172 ] = None,
173 workflow: Annotated[
174 Optional[str],
175 typer.Option(
176 "--workflow",
177 "-w",
178 help="The name of the workflow to run. If not provided, the default workflow will be run.",
179 ),
180 ] = None,
181 single_run: Annotated[
182 bool,
183 typer.Option(
184 "--single-run", help="If set, only a single workflow will be executed."
185 ),
186 ] = False,
187 ctx: typer.Context,
188) -> None:
189 dev_mode = True
190 if len(ctx.args) > 0:
191 raise NotImplementedError("Extra arguments are not supported in this command.")
192 else:
193 params: dict[str, Any] = {}
194 _run_app(
195 path=path,
196 app=app,
197 workflow=workflow,
198 params=params,
199 dev_mode=dev_mode,
200 single_run=single_run,
201 )
204@app.command(help="Display the version of FastAgency") 1afghbicde
205def version() -> None: 1afghbicde
206 typer.echo(__version__)
209def main() -> None: 1afghbicde
210 app()