Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 23 additions & 30 deletions src/fastapi_cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,19 +146,19 @@ def _run(
with get_rich_toolkit() as toolkit:
server_type = "development" if command == "dev" else "production"

toolkit.print_title(f"Starting {server_type} server 🚀", tag="FastAPI")
toolkit.print_line()
toolkit.print(f"Starting FastAPI in {server_type} mode", emoji="⚡️")

toolkit.print_line()
toolkit.print(
"Searching for package file structure from directories with [blue]__init__.py[/blue] files"
"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"
)
toolkit.print_line()
raise typer.Exit(code=1)

try:
Expand All @@ -172,8 +172,6 @@ def _run(
field = ".".join(str(loc) for loc in error["loc"])
toolkit.print(f" [red]•[/red] {field}: {error['msg']}")

toolkit.print_line()

raise typer.Exit(code=1) from None

try:
Expand All @@ -197,45 +195,43 @@ def _run(
module_data = import_data.module_data
import_string = import_data.import_string

toolkit.print(f"Importing from {module_data.extra_sys_path}")
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, tag="module")
toolkit.print_line()
toolkit.print(root_tree)

toolkit.print_line()

toolkit.print(
"Importing the FastAPI app object from the module with the following code:",
tag="code",
)
toolkit.print_line()
toolkit.print(
f"[underline]from [bold]{module_data.module_import_str}[/bold] import [bold]{import_data.app_name}[/bold]"
f"[blue]from [bold]{module_data.module_import_str}[/bold] import [bold]{import_data.app_name}[/bold]",
)
toolkit.print_line()

toolkit.print(
f"Using import string: [blue]{import_string}[/]",
tag="app",
)
toolkit.print_line()
toolkit.print(f"Using import string: [blue]{import_string}[/]", emoji="🐍")

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_line()
toolkit.print("Configuration sources:", tag="info")
toolkit.print("Configuration sources:", emoji="📋")
if mod_source_desc == app_source_desc:
toolkit.print(f" • Import string: {mod_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}")
toolkit.print(f"• Module: {mod_source_desc}")
toolkit.print(f"• App name: {app_source_desc}")

if import_data.module_config_source == "auto-discovery":
toolkit.print_line()
toolkit.print(
"You can configure an entrypoint in [blue]pyproject.toml[/] for this app with:",
tag="tip",
emoji="💡",
)
toolkit.print_line()
toolkit.print(
Expand All @@ -246,24 +242,21 @@ def _run(
),
"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}[/]",
f"Documentation at [link={url_docs}]{url_docs}[/]",
tag="server",
)
toolkit.print(f"Server started at [link={url}]{url}[/]", emoji="🌐")
toolkit.print(f"Documentation at [link={url_docs}]{url_docs}[/]")

if command == "dev":
toolkit.print_line()
toolkit.print(
"Running in development mode, for production use: [bold]fastapi run[/]",
tag="tip",
emoji="💡",
)

if not uvicorn:
Expand All @@ -272,7 +265,7 @@ def _run(
) from None

toolkit.print_line()
toolkit.print("Logs:")
toolkit.print("Logs:", bullet=False)
toolkit.print_line()

extra_uvicorn_kwargs: dict[str, Any] = (
Expand Down
153 changes: 126 additions & 27 deletions src/fastapi_cli/utils/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,130 @@
import sys
from typing import Any

from rich_toolkit import RichToolkit, RichToolkitTheme
from rich_toolkit.styles import TaggedStyle
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:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

these are from fastapi-cloud-cli, ideally in future we'll remove them from fastapi-cloud-cli 😊

"""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:
Expand All @@ -14,18 +134,15 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:

def formatMessage(self, record: logging.LogRecord) -> str:
message = record.getMessage()
result = self.toolkit.print_as_string(message, tag=record.levelname)
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 should_use_rich_logs() -> bool:
"""Return True when stdout is a TTY and rich logs should be used, False otherwise."""
return sys.stdout.isatty()


def get_uvicorn_log_config() -> dict[str, Any]:
return {
"version": 1,
Expand Down Expand Up @@ -69,23 +186,5 @@ def get_uvicorn_log_config() -> dict[str, Any]:
}


logger = logging.getLogger(__name__)


def get_rich_toolkit() -> RichToolkit:
theme = RichToolkitTheme(
style=TaggedStyle(tag_width=11),
theme={
"tag.title": "white on #009485",
"tag": "white on #007166",
"placeholder": "grey85",
"text": "white",
"selected": "#007166",
"result": "grey85",
"progress": "on #007166",
"error": "red",
"log.info": "black on blue",
},
)

return RichToolkit(theme=theme)
return RichToolkit(style=FastAPIStyle())
Loading
Loading