feat: simplify CLI feature (#23)

* chore: cleanup + add tests

* chore: remove graph and show fixtures commands

* feat: add async_command wrapper
This commit is contained in:
d3vyce
2026-02-03 14:35:15 +01:00
committed by GitHub
parent 8c287b3ce7
commit 34ef4da317
7 changed files with 492 additions and 228 deletions

View File

@@ -59,7 +59,7 @@ dev = [
]
[project.scripts]
fastapi-toolsets = "fastapi_toolsets.cli:app"
manager = "fastapi_toolsets.cli:cli"
[build-system]
requires = ["uv_build>=0.9.26,<0.10.0"]

View File

@@ -1,5 +1,6 @@
"""CLI for FastAPI projects."""
from .app import app, register_command
from .app import cli
from .utils import async_command
__all__ = ["app", "register_command"]
__all__ = ["async_command", "cli"]

View File

@@ -1,97 +1,25 @@
"""Main CLI application."""
import importlib.util
import sys
from pathlib import Path
from typing import Annotated
import typer
from .commands import fixtures
from .config import load_config
app = typer.Typer(
name="fastapi-utils",
cli = typer.Typer(
name="manager",
help="CLI utilities for FastAPI projects.",
no_args_is_help=True,
)
# Register built-in commands
app.add_typer(fixtures.app, name="fixtures")
_config = load_config()
if _config.fixtures:
from .commands.fixtures import fixture_cli
cli.add_typer(fixture_cli, name="fixtures")
def register_command(command: typer.Typer, name: str) -> None:
"""Register a custom command group.
Args:
command: Typer app for the command group
name: Name for the command group
Example:
# In your project's cli.py:
import typer
from fastapi_toolsets.cli import app, register_command
my_commands = typer.Typer()
@my_commands.command()
def seed():
'''Seed the database.'''
...
register_command(my_commands, "db")
# Now available as: fastapi-utils db seed
"""
app.add_typer(command, name=name)
@app.callback()
def main(
ctx: typer.Context,
config: Annotated[
Path | None,
typer.Option(
"--config",
"-c",
help="Path to project config file (Python module with fixtures registry).",
envvar="FASTAPI_TOOLSETS_CONFIG",
),
] = None,
) -> None:
@cli.callback()
def main(ctx: typer.Context) -> None:
"""FastAPI utilities CLI."""
ctx.ensure_object(dict)
if config:
ctx.obj["config_path"] = config
# Load the config module
config_module = _load_module_from_path(config)
ctx.obj["config_module"] = config_module
def _load_module_from_path(path: Path) -> object:
"""Load a Python module from a file path.
Handles both absolute and relative imports by adding the config's
parent directory to sys.path temporarily.
"""
path = path.resolve()
# Add the parent directory to sys.path to support relative imports
parent_dir = str(
path.parent.parent
) # Go up two levels (e.g., from app/cli_config.py to project root)
if parent_dir not in sys.path:
sys.path.insert(0, parent_dir)
# Also add immediate parent for direct module imports
immediate_parent = str(path.parent)
if immediate_parent not in sys.path:
sys.path.insert(0, immediate_parent)
spec = importlib.util.spec_from_file_location("config", path)
if spec is None or spec.loader is None:
raise typer.BadParameter(f"Cannot load module from {path}")
module = importlib.util.module_from_spec(spec)
sys.modules["config"] = module
spec.loader.exec_module(module)
return module
ctx.obj["config"] = _config

View File

@@ -1,55 +1,29 @@
"""Fixture management commands."""
import asyncio
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from ...fixtures import Context, FixtureRegistry, LoadStrategy, load_fixtures_by_context
from ...fixtures import Context, LoadStrategy, load_fixtures_by_context
from ..config import CliConfig
from ..utils import async_command
app = typer.Typer(
fixture_cli = typer.Typer(
name="fixtures",
help="Manage database fixtures.",
no_args_is_help=True,
)
console = Console()
def _get_registry(ctx: typer.Context) -> FixtureRegistry:
"""Get fixture registry from context."""
config = ctx.obj.get("config_module") if ctx.obj else None
if config is None:
raise typer.BadParameter(
"No config provided. Use --config to specify a config file with a 'fixtures' registry."
)
registry = getattr(config, "fixtures", None)
if registry is None:
raise typer.BadParameter(
"Config module must have a 'fixtures' attribute (FixtureRegistry instance)."
)
if not isinstance(registry, FixtureRegistry):
raise typer.BadParameter(
f"'fixtures' must be a FixtureRegistry instance, got {type(registry).__name__}"
)
return registry
def _get_config(ctx: typer.Context) -> CliConfig:
"""Get CLI config from context."""
return ctx.obj["config"]
def _get_db_context(ctx: typer.Context):
"""Get database context manager from config."""
config = ctx.obj.get("config_module") if ctx.obj else None
if config is None:
raise typer.BadParameter("No config provided.")
get_db_context = getattr(config, "get_db_context", None)
if get_db_context is None:
raise typer.BadParameter("Config module must have a 'get_db_context' function.")
return get_db_context
@app.command("list")
@fixture_cli.command("list")
def list_fixtures(
ctx: typer.Context,
context: Annotated[
@@ -62,64 +36,28 @@ def list_fixtures(
] = None,
) -> None:
"""List all registered fixtures."""
registry = _get_registry(ctx)
if context:
fixtures = registry.get_by_context(context)
else:
fixtures = registry.get_all()
config = _get_config(ctx)
registry = config.get_fixtures_registry()
fixtures = registry.get_by_context(context) if context else registry.get_all()
if not fixtures:
typer.echo("No fixtures found.")
print("No fixtures found.")
return
typer.echo(f"\n{'Name':<30} {'Contexts':<30} {'Dependencies'}")
typer.echo("-" * 80)
table = Table("Name", "Contexts", "Dependencies")
for fixture in fixtures:
contexts = ", ".join(fixture.contexts)
deps = ", ".join(fixture.depends_on) if fixture.depends_on else "-"
typer.echo(f"{fixture.name:<30} {contexts:<30} {deps}")
table.add_row(fixture.name, contexts, deps)
typer.echo(f"\nTotal: {len(fixtures)} fixture(s)")
console.print(table)
print(f"\nTotal: {len(fixtures)} fixture(s)")
@app.command("graph")
def show_graph(
ctx: typer.Context,
fixture_name: Annotated[
str | None,
typer.Argument(help="Show dependencies for a specific fixture."),
] = None,
) -> None:
"""Show fixture dependency graph."""
registry = _get_registry(ctx)
if fixture_name:
try:
order = registry.resolve_dependencies(fixture_name)
typer.echo(f"\nDependency chain for '{fixture_name}':\n")
for i, name in enumerate(order):
indent = " " * i
arrow = "└─> " if i > 0 else ""
typer.echo(f"{indent}{arrow}{name}")
except KeyError:
typer.echo(f"Fixture '{fixture_name}' not found.", err=True)
raise typer.Exit(1)
else:
# Show full graph
fixtures = registry.get_all()
typer.echo("\nFixture Dependency Graph:\n")
for fixture in fixtures:
deps = (
f" -> [{', '.join(fixture.depends_on)}]" if fixture.depends_on else ""
)
typer.echo(f" {fixture.name}{deps}")
@app.command("load")
def load(
@fixture_cli.command("load")
@async_command
async def load(
ctx: typer.Context,
contexts: Annotated[
list[str] | None,
@@ -141,16 +79,12 @@ def load(
] = False,
) -> None:
"""Load fixtures into the database."""
registry = _get_registry(ctx)
get_db_context = _get_db_context(ctx)
config = _get_config(ctx)
registry = config.get_fixtures_registry()
get_db_context = config.get_db_context()
# Parse contexts
if contexts:
context_list = contexts
else:
context_list = [Context.BASE]
context_list = contexts if contexts else [Context.BASE]
# Parse strategy
try:
load_strategy = LoadStrategy(strategy)
except ValueError:
@@ -159,67 +93,27 @@ def load(
)
raise typer.Exit(1)
# Resolve what will be loaded
ordered = registry.resolve_context_dependencies(*context_list)
if not ordered:
typer.echo("No fixtures to load for the specified context(s).")
print("No fixtures to load for the specified context(s).")
return
typer.echo(f"\nFixtures to load ({load_strategy.value} strategy):")
print(f"\nFixtures to load ({load_strategy.value} strategy):")
for name in ordered:
fixture = registry.get(name)
instances = list(fixture.func())
model_name = type(instances[0]).__name__ if instances else "?"
typer.echo(f" - {name}: {len(instances)} {model_name}(s)")
print(f" - {name}: {len(instances)} {model_name}(s)")
if dry_run:
typer.echo("\n[Dry run - no changes made]")
print("\n[Dry run - no changes made]")
return
typer.echo("\nLoading...")
async def do_load():
async with get_db_context() as session:
result = await load_fixtures_by_context(
session, registry, *context_list, strategy=load_strategy
)
return result
result = asyncio.run(do_load())
async with get_db_context() as session:
result = await load_fixtures_by_context(
session, registry, *context_list, strategy=load_strategy
)
total = sum(len(items) for items in result.values())
typer.echo(f"\nLoaded {total} record(s) successfully.")
@app.command("show")
def show_fixture(
ctx: typer.Context,
name: Annotated[str, typer.Argument(help="Fixture name to show.")],
) -> None:
"""Show details of a specific fixture."""
registry = _get_registry(ctx)
try:
fixture = registry.get(name)
except KeyError:
typer.echo(f"Fixture '{name}' not found.", err=True)
raise typer.Exit(1)
typer.echo(f"\nFixture: {fixture.name}")
typer.echo(f"Contexts: {', '.join(fixture.contexts)}")
typer.echo(
f"Dependencies: {', '.join(fixture.depends_on) if fixture.depends_on else 'None'}"
)
# Show instances
instances = list(fixture.func())
if instances:
model_name = type(instances[0]).__name__
typer.echo(f"\nInstances ({len(instances)} {model_name}):")
for instance in instances[:10]: # Limit to 10
typer.echo(f" - {instance!r}")
if len(instances) > 10:
typer.echo(f" ... and {len(instances) - 10} more")
else:
typer.echo("\nNo instances (empty fixture)")
print(f"\nLoaded {total} record(s) successfully.")

View File

@@ -0,0 +1,92 @@
"""CLI configuration."""
import importlib
import sys
import tomllib
from dataclasses import dataclass
from pathlib import Path
import typer
@dataclass
class CliConfig:
"""CLI configuration loaded from pyproject.toml."""
fixtures: str | None = None
db_context: str | None = None
def get_fixtures_registry(self):
"""Import and return the fixtures registry."""
from ..fixtures import FixtureRegistry
if not self.fixtures:
raise typer.BadParameter(
"No fixtures registry configured. "
"Add 'fixtures' to [tool.fastapi-toolsets] in pyproject.toml."
)
registry = _import_from_string(self.fixtures)
if not isinstance(registry, FixtureRegistry):
raise typer.BadParameter(
f"'fixtures' must be a FixtureRegistry instance, got {type(registry).__name__}"
)
return registry
def get_db_context(self):
"""Import and return the db_context function."""
if not self.db_context:
raise typer.BadParameter(
"No db_context configured. "
"Add 'db_context' to [tool.fastapi-toolsets] in pyproject.toml."
)
return _import_from_string(self.db_context)
def _import_from_string(import_path: str):
"""Import an object from a string path like 'module.submodule:attribute'."""
if ":" not in import_path:
raise typer.BadParameter(
f"Invalid import path '{import_path}'. Expected format: 'module:attribute'"
)
module_path, attr_name = import_path.rsplit(":", 1)
# Add cwd to sys.path for local imports
cwd = str(Path.cwd())
if cwd not in sys.path:
sys.path.insert(0, cwd)
try:
module = importlib.import_module(module_path)
except ImportError as e:
raise typer.BadParameter(f"Cannot import module '{module_path}': {e}")
if not hasattr(module, attr_name):
raise typer.BadParameter(
f"Module '{module_path}' has no attribute '{attr_name}'"
)
return getattr(module, attr_name)
def load_config() -> CliConfig:
"""Load CLI configuration from pyproject.toml."""
pyproject_path = Path.cwd() / "pyproject.toml"
if not pyproject_path.exists():
return CliConfig()
try:
with open(pyproject_path, "rb") as f:
data = tomllib.load(f)
tool_config = data.get("tool", {}).get("fastapi-toolsets", {})
return CliConfig(
fixtures=tool_config.get("fixtures"),
db_context=tool_config.get("db_context"),
)
except Exception:
return CliConfig()

View File

@@ -0,0 +1,27 @@
"""CLI utility functions."""
import asyncio
import functools
from collections.abc import Callable, Coroutine
from typing import Any, ParamSpec, TypeVar
P = ParamSpec("P")
T = TypeVar("T")
def async_command(func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, T]:
"""Decorator to run an async function as a sync CLI command.
Example:
@fixture_cli.command("load")
@async_command
async def load(ctx: typer.Context) -> None:
async with get_db_context() as session:
await load_fixtures(session, registry)
"""
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
return asyncio.run(func(*args, **kwargs))
return wrapper

322
tests/test_cli.py Normal file
View File

@@ -0,0 +1,322 @@
"""Tests for fastapi_toolsets.cli module."""
import sys
import pytest
from typer.testing import CliRunner
from fastapi_toolsets.cli.config import CliConfig, _import_from_string, load_config
from fastapi_toolsets.cli.utils import async_command
from fastapi_toolsets.fixtures import FixtureRegistry
runner = CliRunner()
class TestCliConfig:
"""Tests for CliConfig dataclass."""
def test_default_values(self):
"""Config has None defaults."""
config = CliConfig()
assert config.fixtures is None
assert config.db_context is None
def test_with_values(self):
"""Config stores provided values."""
config = CliConfig(
fixtures="app.fixtures:registry",
db_context="app.db:get_session",
)
assert config.fixtures == "app.fixtures:registry"
assert config.db_context == "app.db:get_session"
def test_get_fixtures_registry_without_config(self):
"""get_fixtures_registry raises error when not configured."""
config = CliConfig()
with pytest.raises(Exception) as exc_info:
config.get_fixtures_registry()
assert "No fixtures registry configured" in str(exc_info.value)
def test_get_db_context_without_config(self):
"""get_db_context raises error when not configured."""
config = CliConfig()
with pytest.raises(Exception) as exc_info:
config.get_db_context()
assert "No db_context configured" in str(exc_info.value)
class TestImportFromString:
"""Tests for _import_from_string function."""
def test_import_valid_path(self):
"""Import valid module:attribute path."""
result = _import_from_string("fastapi_toolsets.fixtures:FixtureRegistry")
assert result is FixtureRegistry
def test_import_without_colon_raises_error(self):
"""Import path without colon raises error."""
with pytest.raises(Exception) as exc_info:
_import_from_string("fastapi_toolsets.fixtures.FixtureRegistry")
assert "Expected format: 'module:attribute'" in str(exc_info.value)
def test_import_nonexistent_module_raises_error(self):
"""Import nonexistent module raises error."""
with pytest.raises(Exception) as exc_info:
_import_from_string("nonexistent.module:something")
assert "Cannot import module" in str(exc_info.value)
def test_import_nonexistent_attribute_raises_error(self):
"""Import nonexistent attribute raises error."""
with pytest.raises(Exception) as exc_info:
_import_from_string("fastapi_toolsets.fixtures:NonexistentClass")
assert "has no attribute" in str(exc_info.value)
class TestLoadConfig:
"""Tests for load_config function."""
def test_load_without_pyproject(self, tmp_path, monkeypatch):
"""Returns empty config when no pyproject.toml exists."""
monkeypatch.chdir(tmp_path)
config = load_config()
assert config.fixtures is None
assert config.db_context is None
def test_load_without_tool_section(self, tmp_path, monkeypatch):
"""Returns empty config when no [tool.fastapi-toolsets] section."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text("[project]\nname = 'test'\n")
monkeypatch.chdir(tmp_path)
config = load_config()
assert config.fixtures is None
assert config.db_context is None
def test_load_with_fixtures_config(self, tmp_path, monkeypatch):
"""Loads fixtures config from pyproject.toml."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
'[tool.fastapi-toolsets]\nfixtures = "app.fixtures:registry"\n'
)
monkeypatch.chdir(tmp_path)
config = load_config()
assert config.fixtures == "app.fixtures:registry"
assert config.db_context is None
def test_load_with_full_config(self, tmp_path, monkeypatch):
"""Loads full config from pyproject.toml."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
"[tool.fastapi-toolsets]\n"
'fixtures = "app.fixtures:registry"\n'
'db_context = "app.db:get_session"\n'
)
monkeypatch.chdir(tmp_path)
config = load_config()
assert config.fixtures == "app.fixtures:registry"
assert config.db_context == "app.db:get_session"
def test_load_with_invalid_toml(self, tmp_path, monkeypatch):
"""Returns empty config when pyproject.toml is invalid."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text("invalid toml {{{")
monkeypatch.chdir(tmp_path)
config = load_config()
assert config.fixtures is None
class TestCliApp:
"""Tests for CLI application."""
def test_cli_import(self):
"""CLI can be imported."""
from fastapi_toolsets.cli import cli
assert cli is not None
def test_cli_help(self, tmp_path, monkeypatch):
"""CLI shows help without fixtures."""
monkeypatch.chdir(tmp_path)
# Need to reload the module to pick up new cwd
import importlib
from fastapi_toolsets.cli import app
importlib.reload(app)
result = runner.invoke(app.cli, ["--help"])
assert result.exit_code == 0
assert "CLI utilities for FastAPI projects" in result.output
class TestFixturesCli:
"""Tests for fixtures CLI commands."""
@pytest.fixture
def cli_env(self, tmp_path, monkeypatch):
"""Set up CLI environment with fixtures config."""
# Create pyproject.toml
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
"[tool.fastapi-toolsets]\n"
'fixtures = "fixtures:registry"\n'
'db_context = "db:get_session"\n'
)
# Create fixtures module
fixtures_file = tmp_path / "fixtures.py"
fixtures_file.write_text(
"from fastapi_toolsets.fixtures import FixtureRegistry, Context\n"
"\n"
"registry = FixtureRegistry()\n"
"\n"
"@registry.register(contexts=[Context.BASE])\n"
"def roles():\n"
' return [{"id": 1, "name": "admin"}, {"id": 2, "name": "user"}]\n'
"\n"
'@registry.register(depends_on=["roles"], contexts=[Context.TESTING])\n'
"def users():\n"
' return [{"id": 1, "name": "alice", "role_id": 1}]\n'
)
# Create db module
db_file = tmp_path / "db.py"
db_file.write_text(
"from contextlib import asynccontextmanager\n"
"\n"
"@asynccontextmanager\n"
"async def get_session():\n"
" yield None\n"
)
monkeypatch.chdir(tmp_path)
# Add tmp_path to sys.path for imports
if str(tmp_path) not in sys.path:
sys.path.insert(0, str(tmp_path))
# Reload the CLI module to pick up new config
import importlib
from fastapi_toolsets.cli import app, config
importlib.reload(config)
importlib.reload(app)
yield tmp_path, app.cli
# Cleanup
if str(tmp_path) in sys.path:
sys.path.remove(str(tmp_path))
def test_fixtures_list(self, cli_env):
"""fixtures list shows registered fixtures."""
tmp_path, cli = cli_env
result = runner.invoke(cli, ["fixtures", "list"])
assert result.exit_code == 0
assert "roles" in result.output
assert "users" in result.output
assert "Total: 2 fixture(s)" in result.output
def test_fixtures_list_with_context(self, cli_env):
"""fixtures list --context filters by context."""
tmp_path, cli = cli_env
result = runner.invoke(cli, ["fixtures", "list", "--context", "base"])
assert result.exit_code == 0
assert "roles" in result.output
assert "users" not in result.output
assert "Total: 1 fixture(s)" in result.output
def test_fixtures_load_dry_run(self, cli_env):
"""fixtures load --dry-run shows what would be loaded."""
tmp_path, cli = cli_env
result = runner.invoke(cli, ["fixtures", "load", "base", "--dry-run"])
assert result.exit_code == 0
assert "Fixtures to load" in result.output
assert "roles" in result.output
assert "[Dry run - no changes made]" in result.output
def test_fixtures_load_invalid_strategy(self, cli_env):
"""fixtures load with invalid strategy shows error."""
tmp_path, cli = cli_env
result = runner.invoke(
cli, ["fixtures", "load", "base", "--strategy", "invalid"]
)
assert result.exit_code == 1
assert "Invalid strategy" in result.output
class TestCliWithoutFixturesConfig:
"""Tests for CLI when fixtures is not configured."""
def test_no_fixtures_command(self, tmp_path, monkeypatch):
"""fixtures command is not available when not configured."""
# Create pyproject.toml without fixtures
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text('[project]\nname = "test"\n')
monkeypatch.chdir(tmp_path)
# Reload the CLI module
import importlib
from fastapi_toolsets.cli import app, config
importlib.reload(config)
importlib.reload(app)
result = runner.invoke(app.cli, ["--help"])
assert result.exit_code == 0
assert "fixtures" not in result.output
class TestAsyncCommand:
"""Tests for async_command decorator."""
def test_async_command_runs_coroutine(self):
"""async_command runs async function synchronously."""
@async_command
async def async_func(value: int) -> int:
return value * 2
result = async_func(21)
assert result == 42
def test_async_command_preserves_signature(self):
"""async_command preserves function signature."""
@async_command
async def async_func(name: str, count: int = 1) -> str:
return f"{name} x {count}"
result = async_func("test", count=3)
assert result == "test x 3"
def test_async_command_preserves_docstring(self):
"""async_command preserves function docstring."""
@async_command
async def async_func() -> None:
"""This is a docstring."""
pass
assert async_func.__doc__ == """This is a docstring."""
def test_async_command_preserves_name(self):
"""async_command preserves function name."""
@async_command
async def my_async_function() -> None:
pass
assert my_async_function.__name__ == "my_async_function"