fix: flatten React Router structure — no nested Routes, each page is a top-level route
This commit is contained in:
enzo
@@ -0,0 +1 @@
|
||||
__version__ = "0.0.32"
|
||||
@@ -0,0 +1,3 @@
|
||||
from .cli import main
|
||||
|
||||
main()
|
||||
@@ -0,0 +1,539 @@
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Any
|
||||
|
||||
import typer
|
||||
from pydantic import ValidationError
|
||||
from rich import print
|
||||
from rich.syntax import Syntax
|
||||
from rich.tree import Tree
|
||||
|
||||
from fastapi_cli.config import FastAPIConfig
|
||||
from fastapi_cli.discover import (
|
||||
AppConfigSource,
|
||||
ModuleConfigSource,
|
||||
get_import_data,
|
||||
get_import_data_from_import_string,
|
||||
)
|
||||
from fastapi_cli.exceptions import FastAPICLIException
|
||||
|
||||
from . import __version__
|
||||
from .logging import setup_logging
|
||||
from .utils.cli import (
|
||||
get_rich_toolkit,
|
||||
get_uvicorn_log_config,
|
||||
should_use_rich_logs,
|
||||
)
|
||||
|
||||
app = typer.Typer(
|
||||
rich_markup_mode="rich", context_settings={"help_option_names": ["-h", "--help"]}
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
SOURCE_DESCRIPTIONS: dict[ModuleConfigSource | AppConfigSource, str] = {
|
||||
"entrypoint-option": "[blue]--entrypoint[/] CLI option",
|
||||
"entrypoint-pyproject": "[blue]entrypoint[/] in [blue]pyproject.toml[/]",
|
||||
"path-argument": "[blue]path[/] CLI argument",
|
||||
"app-option": "[blue]--app[/] CLI option",
|
||||
"auto-discovery": "auto-discovery",
|
||||
}
|
||||
|
||||
|
||||
try:
|
||||
import uvicorn
|
||||
except ImportError: # pragma: no cover
|
||||
uvicorn = None # type: ignore[assignment] # ty: ignore[invalid-assignment]
|
||||
|
||||
|
||||
try:
|
||||
from fastapi_cloud_cli.cli import (
|
||||
app as fastapi_cloud_cli,
|
||||
)
|
||||
|
||||
app.add_typer(fastapi_cloud_cli)
|
||||
except ImportError: # pragma: no cover
|
||||
pass
|
||||
|
||||
|
||||
try:
|
||||
from fastapi_new.cli import ( # type: ignore[import-not-found] # ty: ignore[unresolved-import]
|
||||
app as fastapi_new_cli,
|
||||
)
|
||||
|
||||
app.add_typer(fastapi_new_cli) # pragma: no cover
|
||||
except ImportError: # pragma: no cover
|
||||
pass
|
||||
|
||||
|
||||
def version_callback(value: bool) -> None:
|
||||
if value:
|
||||
print(f"FastAPI CLI version: [green]{__version__}[/green]")
|
||||
try:
|
||||
from fastapi_cloud_cli import (
|
||||
__version__ as cloud_cli_version,
|
||||
)
|
||||
|
||||
print(f"FastAPI Cloud CLI version: [green]{cloud_cli_version}[/green]")
|
||||
except ImportError: # pragma: no cover
|
||||
pass
|
||||
|
||||
raise typer.Exit()
|
||||
|
||||
|
||||
@app.callback()
|
||||
def callback(
|
||||
version: Annotated[
|
||||
bool | None,
|
||||
typer.Option(
|
||||
"--version", help="Show the version and exit.", callback=version_callback
|
||||
),
|
||||
] = None,
|
||||
verbose: bool = typer.Option(False, help="Enable verbose output"),
|
||||
) -> None:
|
||||
"""
|
||||
FastAPI CLI - The [bold]fastapi[/bold] command line app. 😎
|
||||
|
||||
Manage your [bold]FastAPI[/bold] projects, run your FastAPI apps, and more.
|
||||
|
||||
Read more in the docs: [link=https://fastapi.tiangolo.com/fastapi-cli/]https://fastapi.tiangolo.com/fastapi-cli/[/link].
|
||||
"""
|
||||
|
||||
log_level = logging.DEBUG if verbose else logging.INFO
|
||||
|
||||
setup_logging(level=log_level)
|
||||
|
||||
|
||||
def _get_module_tree(module_paths: list[Path]) -> Tree:
|
||||
root = module_paths[0]
|
||||
name = f"🐍 {root.name}" if root.is_file() else f"📁 {root.name}"
|
||||
|
||||
root_tree = Tree(name)
|
||||
|
||||
if root.is_dir():
|
||||
root_tree.add("[dim]🐍 __init__.py[/dim]")
|
||||
|
||||
tree = root_tree
|
||||
for sub_path in module_paths[1:]:
|
||||
sub_name = (
|
||||
f"🐍 {sub_path.name}" if sub_path.is_file() else f"📁 {sub_path.name}"
|
||||
)
|
||||
tree = tree.add(sub_name)
|
||||
if sub_path.is_dir():
|
||||
tree.add("[dim]🐍 __init__.py[/dim]")
|
||||
|
||||
return root_tree
|
||||
|
||||
|
||||
def _run(
|
||||
path: Path | None = None,
|
||||
*,
|
||||
host: str = "127.0.0.1",
|
||||
port: int = 8000,
|
||||
reload: bool = True,
|
||||
reload_dirs: list[Path] | None = None,
|
||||
workers: int | None = None,
|
||||
root_path: str = "",
|
||||
command: str,
|
||||
app: str | None = None,
|
||||
entrypoint: str | None = None,
|
||||
proxy_headers: bool = False,
|
||||
forwarded_allow_ips: str | None = None,
|
||||
public_url: str | None = None,
|
||||
verbose: bool = False,
|
||||
) -> None:
|
||||
with get_rich_toolkit() as toolkit:
|
||||
server_type = "development" if command == "dev" else "production"
|
||||
|
||||
toolkit.print(f"Starting FastAPI in {server_type} mode", emoji="⚡️")
|
||||
|
||||
if verbose:
|
||||
toolkit.print_line()
|
||||
toolkit.print(
|
||||
"Searching for package file structure from directories with [blue]__init__.py[/blue] files",
|
||||
emoji="🔎",
|
||||
)
|
||||
|
||||
if entrypoint and (path or app):
|
||||
toolkit.print_line()
|
||||
toolkit.print(
|
||||
"[error]Cannot use --entrypoint together with path or --app arguments"
|
||||
)
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
try:
|
||||
config = FastAPIConfig.resolve(entrypoint=entrypoint)
|
||||
except ValidationError as e:
|
||||
toolkit.print_line()
|
||||
toolkit.print("[error]Invalid configuration in pyproject.toml:")
|
||||
toolkit.print_line()
|
||||
|
||||
for error in e.errors():
|
||||
field = ".".join(str(loc) for loc in error["loc"])
|
||||
toolkit.print(f" [red]•[/red] {field}: {error['msg']}")
|
||||
|
||||
raise typer.Exit(code=1) from None
|
||||
|
||||
try:
|
||||
# Resolve import data with priority: CLI path/app > config entrypoint > auto-discovery
|
||||
if path or app:
|
||||
import_data = get_import_data(path=path, app_name=app)
|
||||
elif config.entrypoint:
|
||||
import_data = get_import_data_from_import_string(
|
||||
config.entrypoint, config.from_pyproject
|
||||
)
|
||||
else:
|
||||
import_data = get_import_data()
|
||||
except FastAPICLIException as e:
|
||||
toolkit.print_line()
|
||||
toolkit.print(f"[error]{e}")
|
||||
raise typer.Exit(code=1) from None
|
||||
|
||||
logger.debug(f"Importing from {import_data.module_data.extra_sys_path}")
|
||||
logger.debug(f"Importing module {import_data.module_data.module_import_str}")
|
||||
|
||||
module_data = import_data.module_data
|
||||
import_string = import_data.import_string
|
||||
is_auto_discovery = import_data.module_config_source == "auto-discovery"
|
||||
|
||||
if verbose:
|
||||
toolkit.print_line()
|
||||
toolkit.print(f"Importing from {module_data.extra_sys_path}", emoji="📂")
|
||||
toolkit.print_line()
|
||||
|
||||
if module_data.module_paths:
|
||||
root_tree = _get_module_tree(module_data.module_paths)
|
||||
|
||||
toolkit.print(root_tree)
|
||||
|
||||
toolkit.print_line()
|
||||
|
||||
toolkit.print(
|
||||
"Importing the FastAPI app object from the module with the following code:",
|
||||
)
|
||||
toolkit.print_line()
|
||||
toolkit.print(
|
||||
f"[blue]from [bold]{module_data.module_import_str}[/bold] import [bold]{import_data.app_name}[/bold]",
|
||||
)
|
||||
|
||||
# Outside --verbose, fold the resolution source into the import string
|
||||
# line and point newcomers at --verbose for the full explanation
|
||||
if is_auto_discovery and not verbose:
|
||||
app_note = " [dim](auto-discovered, use --verbose to learn more)[/]"
|
||||
else:
|
||||
app_note = ""
|
||||
|
||||
toolkit.print_line()
|
||||
toolkit.print(
|
||||
f"Using import string: [blue]{import_string}[/]{app_note}", emoji="🐍"
|
||||
)
|
||||
|
||||
if verbose:
|
||||
toolkit.print_line()
|
||||
mod_source_desc = SOURCE_DESCRIPTIONS[import_data.module_config_source]
|
||||
app_source_desc = SOURCE_DESCRIPTIONS[import_data.app_name_config_source]
|
||||
toolkit.print("Configuration sources:", emoji="📋")
|
||||
if mod_source_desc == app_source_desc:
|
||||
toolkit.print(f"• Import string: {mod_source_desc}")
|
||||
else:
|
||||
toolkit.print(f"• Module: {mod_source_desc}")
|
||||
toolkit.print(f"• App name: {app_source_desc}")
|
||||
|
||||
# Nudge to pin the entrypoint whenever it was auto-discovered, so it's
|
||||
# explicit next time — shown in the default output, not just --verbose
|
||||
if is_auto_discovery:
|
||||
toolkit.print_line()
|
||||
toolkit.print(
|
||||
"You can configure an entrypoint in [blue]pyproject.toml[/] for this app with:",
|
||||
emoji="💡",
|
||||
)
|
||||
toolkit.print_line()
|
||||
toolkit.print(
|
||||
Syntax(
|
||||
(
|
||||
"[tool.fastapi]\n"
|
||||
f'entrypoint = "{import_data.module_data.module_import_str}:{import_data.app_name}"'
|
||||
),
|
||||
"toml",
|
||||
theme="ansi_light",
|
||||
),
|
||||
)
|
||||
|
||||
url = public_url.rstrip("/") if public_url else f"http://{host}:{port}"
|
||||
url_docs = f"{url}/docs"
|
||||
|
||||
toolkit.print_line()
|
||||
toolkit.print(f"Server started at [link={url}]{url}[/]", emoji="🌐")
|
||||
toolkit.print(f"Documentation at [link={url_docs}]{url_docs}[/]")
|
||||
|
||||
if not uvicorn:
|
||||
raise FastAPICLIException(
|
||||
"Could not import Uvicorn, try running 'pip install uvicorn'"
|
||||
) from None
|
||||
|
||||
toolkit.print_line()
|
||||
toolkit.print("Logs:", bullet=False)
|
||||
toolkit.print_line()
|
||||
|
||||
extra_uvicorn_kwargs: dict[str, Any] = (
|
||||
{"log_config": get_uvicorn_log_config()} if should_use_rich_logs() else {}
|
||||
)
|
||||
|
||||
uvicorn.run(
|
||||
app=import_string,
|
||||
host=host,
|
||||
port=port,
|
||||
reload=reload,
|
||||
reload_dirs=(
|
||||
[str(directory.resolve()) for directory in reload_dirs]
|
||||
if reload_dirs
|
||||
else None
|
||||
),
|
||||
workers=workers,
|
||||
root_path=root_path,
|
||||
proxy_headers=proxy_headers,
|
||||
forwarded_allow_ips=forwarded_allow_ips,
|
||||
**extra_uvicorn_kwargs,
|
||||
)
|
||||
|
||||
|
||||
@app.command()
|
||||
def dev(
|
||||
path: Annotated[
|
||||
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,
|
||||
*,
|
||||
host: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
help="The host to serve on. For local development in localhost use [blue]127.0.0.1[/blue]. To enable public access, e.g. in a container, use all the IP addresses available with [blue]0.0.0.0[/blue]."
|
||||
),
|
||||
] = "127.0.0.1",
|
||||
port: Annotated[
|
||||
int,
|
||||
typer.Option(
|
||||
help="The port to serve on. You would normally have a termination proxy on top (another program) handling HTTPS on port [blue]443[/blue] and HTTP on port [blue]80[/blue], transferring the communication to your app.",
|
||||
envvar="PORT",
|
||||
),
|
||||
] = 8000,
|
||||
reload: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
help="Enable auto-reload of the server when (code) files change. This is [bold]resource intensive[/bold], use it only during development."
|
||||
),
|
||||
] = True,
|
||||
reload_dir: Annotated[
|
||||
list[Path] | None,
|
||||
typer.Option(
|
||||
help="Set reload directories explicitly, instead of using the current working directory."
|
||||
),
|
||||
] = None,
|
||||
root_path: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
help="The root path is used to tell your app that it is being served to the outside world with some [bold]path prefix[/bold] set up in some termination proxy or similar."
|
||||
),
|
||||
] = "",
|
||||
app: Annotated[
|
||||
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,
|
||||
entrypoint: Annotated[
|
||||
str | None,
|
||||
typer.Option(
|
||||
"--entrypoint",
|
||||
"-e",
|
||||
help="The FastAPI app import string in the format 'some.importable_module:app_name'.",
|
||||
),
|
||||
] = None,
|
||||
proxy_headers: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
help="Enable/Disable X-Forwarded-Proto, X-Forwarded-For, X-Forwarded-Port to populate remote address info."
|
||||
),
|
||||
] = True,
|
||||
forwarded_allow_ips: Annotated[
|
||||
str | None,
|
||||
typer.Option(
|
||||
help="Comma separated list of IP Addresses to trust with proxy headers. The literal '*' means trust everything."
|
||||
),
|
||||
] = None,
|
||||
verbose: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
"--verbose",
|
||||
"-v",
|
||||
help="Show detailed startup output and debug logs: how the [bold]FastAPI[/bold] app was discovered, imported, and configured.",
|
||||
),
|
||||
] = False,
|
||||
) -> Any:
|
||||
"""
|
||||
Run a [bold]FastAPI[/bold] app in [yellow]development[/yellow] mode. 🧪
|
||||
|
||||
This is equivalent to [bold]fastapi run[/bold] but with [bold]reload[/bold] enabled and listening on the [blue]127.0.0.1[/blue] address.
|
||||
|
||||
It automatically detects the Python module or package that needs to be imported based on the file or directory path passed.
|
||||
|
||||
If no path is passed, it tries with:
|
||||
|
||||
- [blue]main.py[/blue]
|
||||
- [blue]app.py[/blue]
|
||||
- [blue]api.py[/blue]
|
||||
- [blue]app/main.py[/blue]
|
||||
- [blue]app/app.py[/blue]
|
||||
- [blue]app/api.py[/blue]
|
||||
|
||||
It also detects the directory that needs to be added to the [bold]PYTHONPATH[/bold] to make the app importable and adds it.
|
||||
|
||||
It detects the [bold]FastAPI[/bold] app object to use. By default it looks in the module or package for an object named:
|
||||
|
||||
- [blue]app[/blue]
|
||||
- [blue]api[/blue]
|
||||
|
||||
Otherwise, it uses the first [bold]FastAPI[/bold] app found in the imported module or package.
|
||||
"""
|
||||
os.environ.setdefault("FASTAPI_ENV", "development")
|
||||
|
||||
_run(
|
||||
path=path,
|
||||
host=host,
|
||||
port=port,
|
||||
reload=reload,
|
||||
reload_dirs=reload_dir,
|
||||
root_path=root_path,
|
||||
app=app,
|
||||
entrypoint=entrypoint,
|
||||
command="dev",
|
||||
proxy_headers=proxy_headers,
|
||||
forwarded_allow_ips=forwarded_allow_ips,
|
||||
public_url=os.getenv("FASTAPI_PUBLIC_URL"),
|
||||
verbose=verbose,
|
||||
)
|
||||
|
||||
|
||||
@app.command()
|
||||
def run(
|
||||
path: Annotated[
|
||||
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,
|
||||
*,
|
||||
host: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
help="The host to serve on. For local development in localhost use [blue]127.0.0.1[/blue]. To enable public access, e.g. in a container, use all the IP addresses available with [blue]0.0.0.0[/blue]."
|
||||
),
|
||||
] = "0.0.0.0",
|
||||
port: Annotated[
|
||||
int,
|
||||
typer.Option(
|
||||
help="The port to serve on. You would normally have a termination proxy on top (another program) handling HTTPS on port [blue]443[/blue] and HTTP on port [blue]80[/blue], transferring the communication to your app.",
|
||||
envvar="PORT",
|
||||
),
|
||||
] = 8000,
|
||||
reload: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
help="Enable auto-reload of the server when (code) files change. This is [bold]resource intensive[/bold], use it only during development."
|
||||
),
|
||||
] = False,
|
||||
workers: Annotated[
|
||||
int | None,
|
||||
typer.Option(
|
||||
help="Use multiple worker processes. Mutually exclusive with the --reload flag."
|
||||
),
|
||||
] = None,
|
||||
root_path: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
help="The root path is used to tell your app that it is being served to the outside world with some [bold]path prefix[/bold] set up in some termination proxy or similar."
|
||||
),
|
||||
] = "",
|
||||
app: Annotated[
|
||||
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,
|
||||
entrypoint: Annotated[
|
||||
str | None,
|
||||
typer.Option(
|
||||
"--entrypoint",
|
||||
"-e",
|
||||
help="The FastAPI app import string in the format 'some.importable_module:app_name'.",
|
||||
),
|
||||
] = None,
|
||||
proxy_headers: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
help="Enable/Disable X-Forwarded-Proto, X-Forwarded-For, X-Forwarded-Port to populate remote address info."
|
||||
),
|
||||
] = True,
|
||||
forwarded_allow_ips: Annotated[
|
||||
str | None,
|
||||
typer.Option(
|
||||
help="Comma separated list of IP Addresses to trust with proxy headers. The literal '*' means trust everything."
|
||||
),
|
||||
] = None,
|
||||
verbose: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
"--verbose",
|
||||
"-v",
|
||||
help="Show detailed startup output and debug logs: how the [bold]FastAPI[/bold] app was discovered, imported, and configured.",
|
||||
),
|
||||
] = False,
|
||||
) -> Any:
|
||||
"""
|
||||
Run a [bold]FastAPI[/bold] app in [green]production[/green] mode. 🚀
|
||||
|
||||
This is equivalent to [bold]fastapi dev[/bold] but with [bold]reload[/bold] disabled and listening on the [blue]0.0.0.0[/blue] address.
|
||||
|
||||
It automatically detects the Python module or package that needs to be imported based on the file or directory path passed.
|
||||
|
||||
If no path is passed, it tries with:
|
||||
|
||||
- [blue]main.py[/blue]
|
||||
- [blue]app.py[/blue]
|
||||
- [blue]api.py[/blue]
|
||||
- [blue]app/main.py[/blue]
|
||||
- [blue]app/app.py[/blue]
|
||||
- [blue]app/api.py[/blue]
|
||||
|
||||
It also detects the directory that needs to be added to the [bold]PYTHONPATH[/bold] to make the app importable and adds it.
|
||||
|
||||
It detects the [bold]FastAPI[/bold] app object to use. By default it looks in the module or package for an object named:
|
||||
|
||||
- [blue]app[/blue]
|
||||
- [blue]api[/blue]
|
||||
|
||||
Otherwise, it uses the first [bold]FastAPI[/bold] app found in the imported module or package.
|
||||
"""
|
||||
_run(
|
||||
path=path,
|
||||
host=host,
|
||||
port=port,
|
||||
reload=reload,
|
||||
workers=workers,
|
||||
root_path=root_path,
|
||||
app=app,
|
||||
entrypoint=entrypoint,
|
||||
command="run",
|
||||
proxy_headers=proxy_headers,
|
||||
forwarded_allow_ips=forwarded_allow_ips,
|
||||
public_url=os.getenv("FASTAPI_PUBLIC_URL"),
|
||||
verbose=verbose,
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
app()
|
||||
@@ -0,0 +1,45 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, StrictStr
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FastAPIConfig(BaseModel):
|
||||
entrypoint: StrictStr | None = None
|
||||
from_pyproject: bool = False
|
||||
|
||||
@classmethod
|
||||
def _read_pyproject_toml(cls) -> dict[str, Any]:
|
||||
"""Read FastAPI configuration from pyproject.toml in current directory."""
|
||||
pyproject_path = Path.cwd() / "pyproject.toml"
|
||||
|
||||
if not pyproject_path.exists():
|
||||
return {}
|
||||
|
||||
try:
|
||||
import tomllib # type: ignore[import-not-found, unused-ignore] # ty: ignore[unresolved-import]
|
||||
except ImportError:
|
||||
try:
|
||||
import tomli as tomllib # type: ignore[no-redef, import-not-found, unused-ignore]
|
||||
except ImportError: # pragma: no cover
|
||||
logger.debug("tomli not available, skipping pyproject.toml")
|
||||
return {}
|
||||
|
||||
with open(pyproject_path, "rb") as f:
|
||||
data = tomllib.load(f)
|
||||
|
||||
return data.get("tool", {}).get("fastapi", {}) # type: ignore[no-any-return]
|
||||
|
||||
@classmethod
|
||||
def resolve(cls, entrypoint: str | None = None) -> "FastAPIConfig":
|
||||
config = cls._read_pyproject_toml()
|
||||
|
||||
if entrypoint is not None:
|
||||
config["entrypoint"] = entrypoint
|
||||
|
||||
config["from_pyproject"] = ("entrypoint" in config) and (entrypoint is None)
|
||||
|
||||
return cls.model_validate(config)
|
||||
@@ -0,0 +1,184 @@
|
||||
import importlib
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from logging import getLogger
|
||||
from pathlib import Path
|
||||
from typing import Literal, TypeAlias
|
||||
|
||||
from fastapi_cli.exceptions import FastAPICLIException
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
try:
|
||||
from fastapi import FastAPI
|
||||
except ImportError: # pragma: no cover
|
||||
FastAPI = None # type: ignore[misc, assignment] # ty: ignore[invalid-assignment]
|
||||
|
||||
|
||||
def get_default_path() -> Path:
|
||||
potential_paths = (
|
||||
"main.py",
|
||||
"app.py",
|
||||
"api.py",
|
||||
"app/main.py",
|
||||
"app/app.py",
|
||||
"app/api.py",
|
||||
)
|
||||
|
||||
for full_path in potential_paths:
|
||||
path = Path(full_path)
|
||||
if path.is_file():
|
||||
return path
|
||||
|
||||
raise FastAPICLIException(
|
||||
"Could not find a default file to run, please provide an explicit path"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ModuleData:
|
||||
module_import_str: str
|
||||
extra_sys_path: Path
|
||||
module_paths: list[Path]
|
||||
|
||||
|
||||
def get_module_data_from_path(path: Path) -> ModuleData:
|
||||
use_path = path.resolve()
|
||||
module_path = use_path
|
||||
if use_path.is_file() and use_path.stem == "__init__":
|
||||
module_path = use_path.parent
|
||||
module_paths = [module_path]
|
||||
extra_sys_path = module_path.parent
|
||||
for parent in module_path.parents:
|
||||
init_path = parent / "__init__.py"
|
||||
if init_path.is_file():
|
||||
module_paths.insert(0, parent)
|
||||
extra_sys_path = parent.parent
|
||||
else:
|
||||
break
|
||||
|
||||
module_str = ".".join(p.stem for p in module_paths)
|
||||
return ModuleData(
|
||||
module_import_str=module_str,
|
||||
extra_sys_path=extra_sys_path.resolve(),
|
||||
module_paths=module_paths,
|
||||
)
|
||||
|
||||
|
||||
def get_app_name(*, mod_data: ModuleData, app_name: str | None = None) -> str:
|
||||
try:
|
||||
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 {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} doesn't seem to be a FastAPI app"
|
||||
)
|
||||
return app_name
|
||||
for preferred_name in ["app", "api"]:
|
||||
if preferred_name in object_names_set:
|
||||
obj = getattr(mod, preferred_name)
|
||||
if isinstance(obj, FastAPI):
|
||||
return preferred_name
|
||||
for name in object_names:
|
||||
obj = getattr(mod, name)
|
||||
if isinstance(obj, FastAPI):
|
||||
return name
|
||||
raise FastAPICLIException("Could not find FastAPI app in module, try using --app")
|
||||
|
||||
|
||||
ModuleConfigSource: TypeAlias = Literal[
|
||||
"entrypoint-option",
|
||||
"entrypoint-pyproject",
|
||||
"path-argument",
|
||||
"auto-discovery",
|
||||
]
|
||||
|
||||
AppConfigSource: TypeAlias = Literal[
|
||||
"entrypoint-option", "entrypoint-pyproject", "app-option", "auto-discovery"
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImportData:
|
||||
app_name: str
|
||||
module_data: ModuleData
|
||||
import_string: str
|
||||
|
||||
module_config_source: ModuleConfigSource
|
||||
app_name_config_source: AppConfigSource
|
||||
|
||||
|
||||
def get_import_data(
|
||||
*, path: Path | None = None, app_name: str | None = None
|
||||
) -> ImportData:
|
||||
path_config_source: ModuleConfigSource = "path-argument"
|
||||
if not path:
|
||||
path = get_default_path()
|
||||
path_config_source = "auto-discovery"
|
||||
|
||||
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)
|
||||
sys.path.insert(0, str(mod_data.extra_sys_path))
|
||||
use_app_name = get_app_name(mod_data=mod_data, app_name=app_name)
|
||||
|
||||
import_string = f"{mod_data.module_import_str}:{use_app_name}"
|
||||
|
||||
return ImportData(
|
||||
app_name=use_app_name,
|
||||
module_data=mod_data,
|
||||
import_string=import_string,
|
||||
module_config_source=path_config_source,
|
||||
app_name_config_source="app-option" if app_name else "auto-discovery",
|
||||
)
|
||||
|
||||
|
||||
def get_import_data_from_import_string(
|
||||
import_string: str, from_pyproject: bool
|
||||
) -> ImportData:
|
||||
module_str, _, app_name = import_string.partition(":")
|
||||
|
||||
if not module_str or not app_name:
|
||||
raise FastAPICLIException(
|
||||
"Import string must be in the format module.submodule:app_name"
|
||||
)
|
||||
|
||||
here = Path(".").resolve()
|
||||
|
||||
sys.path.insert(0, str(here))
|
||||
|
||||
return ImportData(
|
||||
app_name=app_name,
|
||||
module_data=ModuleData(
|
||||
module_import_str=module_str,
|
||||
extra_sys_path=here,
|
||||
module_paths=[],
|
||||
),
|
||||
import_string=import_string,
|
||||
module_config_source=(
|
||||
"entrypoint-pyproject" if from_pyproject else "entrypoint-option"
|
||||
),
|
||||
app_name_config_source=(
|
||||
"entrypoint-pyproject" if from_pyproject else "entrypoint-option"
|
||||
),
|
||||
)
|
||||
@@ -0,0 +1,2 @@
|
||||
class FastAPICLIException(Exception):
|
||||
pass
|
||||
@@ -0,0 +1,22 @@
|
||||
import logging
|
||||
|
||||
from rich.console import Console
|
||||
from rich.logging import RichHandler
|
||||
|
||||
|
||||
def setup_logging(terminal_width: int | None = None, level: int = logging.INFO) -> None:
|
||||
logger = logging.getLogger("fastapi_cli")
|
||||
console = Console(width=terminal_width) if terminal_width else None
|
||||
rich_handler = RichHandler(
|
||||
show_time=False,
|
||||
rich_tracebacks=True,
|
||||
tracebacks_show_locals=True,
|
||||
markup=True,
|
||||
show_path=False,
|
||||
console=console,
|
||||
)
|
||||
rich_handler.setFormatter(logging.Formatter("%(message)s"))
|
||||
logger.addHandler(rich_handler)
|
||||
|
||||
logger.setLevel(level)
|
||||
logger.propagate = False
|
||||
@@ -0,0 +1,190 @@
|
||||
import logging
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from rich._loop import loop_first
|
||||
from rich.console import Console, ConsoleOptions, RenderableType, RenderResult
|
||||
from rich.segment import Segment
|
||||
from rich.text import Text
|
||||
from rich_toolkit import RichToolkit
|
||||
from rich_toolkit.element import Element
|
||||
from rich_toolkit.styles import BaseStyle
|
||||
from uvicorn.logging import DefaultFormatter
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def should_use_rich_logs() -> bool:
|
||||
"""Return True when stdout is a TTY and rich logs should be used, False otherwise."""
|
||||
return sys.stdout.isatty()
|
||||
|
||||
|
||||
class IndentedBlock:
|
||||
"""Indent a renderable, hanging a prefix (e.g. an emoji bullet) on the
|
||||
first line and aligning wrapped/extra lines under the text."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
renderable: RenderableType,
|
||||
*,
|
||||
first_prefix: Text,
|
||||
prefix: Text,
|
||||
) -> None:
|
||||
self.renderable = renderable
|
||||
self.first_prefix = first_prefix
|
||||
self.prefix = prefix
|
||||
|
||||
# Text renders its `end` ("\n" by default), which would break lines
|
||||
self.first_prefix.end = ""
|
||||
self.prefix.end = ""
|
||||
|
||||
def __rich_console__(
|
||||
self, console: Console, options: ConsoleOptions
|
||||
) -> RenderResult:
|
||||
prefix_width = max(self.first_prefix.cell_len, self.prefix.cell_len)
|
||||
lines = console.render_lines(
|
||||
self.renderable,
|
||||
options.update_width(options.max_width - prefix_width),
|
||||
pad=False,
|
||||
)
|
||||
|
||||
new_line = Segment.line()
|
||||
|
||||
for first, line in loop_first(lines):
|
||||
if any(segment.text.strip() for segment in line):
|
||||
yield from console.render(
|
||||
self.first_prefix if first else self.prefix, options
|
||||
)
|
||||
yield from line
|
||||
|
||||
yield new_line
|
||||
|
||||
|
||||
class FastAPIStyle(BaseStyle):
|
||||
"""Uniform left indent with emoji bullets hanging to the left of the text.
|
||||
|
||||
``emoji=`` renders as a bullet in a fixed column; ``bullet=False`` drops
|
||||
the bullet column and just indents. This is a small, purpose-built subset
|
||||
of the richer style in fastapi-cloud-cli.
|
||||
"""
|
||||
|
||||
content_padding = 1
|
||||
emoji_column_width = 3
|
||||
|
||||
def render_element(
|
||||
self,
|
||||
element: Any,
|
||||
is_active: bool = False,
|
||||
done: bool = False,
|
||||
parent: Element | None = None,
|
||||
**kwargs: Any,
|
||||
) -> RenderableType:
|
||||
rendered = super().render_element(
|
||||
element=element, is_active=is_active, done=done, parent=parent, **kwargs
|
||||
)
|
||||
|
||||
emoji = kwargs.get("emoji", "")
|
||||
|
||||
if not emoji and not kwargs.get("bullet", True):
|
||||
indent = Text(" " * (self.content_padding + 1))
|
||||
return IndentedBlock(rendered, first_prefix=indent, prefix=indent)
|
||||
|
||||
return IndentedBlock(
|
||||
rendered,
|
||||
first_prefix=self._get_bullet_prefix(emoji),
|
||||
prefix=Text(" " * (self.content_padding + self.emoji_column_width)),
|
||||
)
|
||||
|
||||
def _get_bullet_prefix(self, emoji: str) -> Text:
|
||||
prefix = Text(" " * self.content_padding)
|
||||
|
||||
if emoji:
|
||||
prefix.append_text(Text.from_markup(emoji))
|
||||
|
||||
prefix.pad_right(
|
||||
self.content_padding + self.emoji_column_width - prefix.cell_len
|
||||
)
|
||||
|
||||
return prefix
|
||||
|
||||
|
||||
LOG_LEVEL_COLORS = {
|
||||
"debug": "blue",
|
||||
"info": "cyan",
|
||||
"warning": "yellow",
|
||||
"warn": "yellow",
|
||||
"error": "red",
|
||||
"critical": "magenta",
|
||||
"fatal": "magenta",
|
||||
}
|
||||
|
||||
|
||||
def _get_log_bullet(level: str) -> str:
|
||||
"""Colored bar rendered in the emoji bullet column, matching the log
|
||||
level, like the ``fastapi cloud logs`` output."""
|
||||
color = LOG_LEVEL_COLORS.get(level.lower(), "dim")
|
||||
|
||||
return f"[{color}]▕[/{color}]"
|
||||
|
||||
|
||||
class CustomFormatter(DefaultFormatter):
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self.toolkit = get_rich_toolkit()
|
||||
|
||||
def formatMessage(self, record: logging.LogRecord) -> str:
|
||||
message = record.getMessage()
|
||||
result = self.toolkit.print_as_string(
|
||||
message, emoji=_get_log_bullet(record.levelname)
|
||||
)
|
||||
# Prepend newline to fix alignment after ^C is printed by the terminal
|
||||
if message == "Shutting down":
|
||||
result = "\n" + result
|
||||
return result
|
||||
|
||||
|
||||
def get_uvicorn_log_config() -> dict[str, Any]:
|
||||
return {
|
||||
"version": 1,
|
||||
"disable_existing_loggers": False,
|
||||
"formatters": {
|
||||
"default": {
|
||||
"()": CustomFormatter,
|
||||
"fmt": "%(levelprefix)s %(message)s",
|
||||
"use_colors": None,
|
||||
},
|
||||
"access": {
|
||||
"()": CustomFormatter,
|
||||
"fmt": "%(levelprefix)s %(client_addr)s - '%(request_line)s' %(status_code)s",
|
||||
},
|
||||
},
|
||||
"handlers": {
|
||||
"default": {
|
||||
"formatter": "default",
|
||||
"class": "logging.StreamHandler",
|
||||
"stream": "ext://sys.stderr",
|
||||
},
|
||||
"access": {
|
||||
"formatter": "access",
|
||||
"class": "logging.StreamHandler",
|
||||
"stream": "ext://sys.stdout",
|
||||
},
|
||||
},
|
||||
"loggers": {
|
||||
"uvicorn": {
|
||||
"handlers": ["default"],
|
||||
"level": "INFO",
|
||||
"propagate": False,
|
||||
},
|
||||
"uvicorn.error": {"level": "INFO"},
|
||||
"uvicorn.access": {
|
||||
"handlers": ["access"],
|
||||
"level": "INFO",
|
||||
"propagate": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def get_rich_toolkit() -> RichToolkit:
|
||||
return RichToolkit(style=FastAPIStyle())
|
||||
Reference in New Issue
Block a user