mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-03-01 17:00:48 +01:00
feat: allow custom CLI (#28)
This commit is contained in:
@@ -1,6 +1,5 @@
|
||||
"""CLI for FastAPI projects."""
|
||||
|
||||
from .app import cli
|
||||
from .utils import async_command
|
||||
|
||||
__all__ = ["async_command", "cli"]
|
||||
__all__ = ["async_command"]
|
||||
|
||||
@@ -2,17 +2,23 @@
|
||||
|
||||
import typer
|
||||
|
||||
from .config import load_config
|
||||
from .config import get_custom_cli
|
||||
from .pyproject import load_pyproject
|
||||
|
||||
cli = typer.Typer(
|
||||
name="manager",
|
||||
help="CLI utilities for FastAPI projects.",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
# Use custom CLI if configured, otherwise create default one
|
||||
_custom_cli = get_custom_cli()
|
||||
|
||||
_config = load_config()
|
||||
if _custom_cli is not None:
|
||||
cli = _custom_cli
|
||||
else:
|
||||
cli = typer.Typer(
|
||||
name="manager",
|
||||
help="CLI utilities for FastAPI projects.",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
|
||||
if _config.fixtures:
|
||||
_config = load_pyproject()
|
||||
if _config.get("fixtures") and _config.get("db_context"):
|
||||
from .commands.fixtures import fixture_cli
|
||||
|
||||
cli.add_typer(fixture_cli, name="fixtures")
|
||||
@@ -22,4 +28,3 @@ if _config.fixtures:
|
||||
def main(ctx: typer.Context) -> None:
|
||||
"""FastAPI utilities CLI."""
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["config"] = _config
|
||||
|
||||
@@ -7,7 +7,7 @@ from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
from ...fixtures import Context, LoadStrategy, load_fixtures_by_context
|
||||
from ..config import CliConfig
|
||||
from ..config import get_db_context, get_fixtures_registry
|
||||
from ..utils import async_command
|
||||
|
||||
fixture_cli = typer.Typer(
|
||||
@@ -18,27 +18,21 @@ fixture_cli = typer.Typer(
|
||||
console = Console()
|
||||
|
||||
|
||||
def _get_config(ctx: typer.Context) -> CliConfig:
|
||||
"""Get CLI config from context."""
|
||||
return ctx.obj["config"]
|
||||
|
||||
|
||||
@fixture_cli.command("list")
|
||||
def list_fixtures(
|
||||
ctx: typer.Context,
|
||||
context: Annotated[
|
||||
str | None,
|
||||
Context | None,
|
||||
typer.Option(
|
||||
"--context",
|
||||
"-c",
|
||||
help="Filter by context (base, production, development, testing).",
|
||||
help="Filter by context.",
|
||||
),
|
||||
] = None,
|
||||
) -> None:
|
||||
"""List all registered fixtures."""
|
||||
config = _get_config(ctx)
|
||||
registry = config.get_fixtures_registry()
|
||||
fixtures = registry.get_by_context(context) if context else registry.get_all()
|
||||
registry = get_fixtures_registry()
|
||||
fixtures = registry.get_by_context(context.value) if context else registry.get_all()
|
||||
|
||||
if not fixtures:
|
||||
print("No fixtures found.")
|
||||
@@ -60,17 +54,13 @@ def list_fixtures(
|
||||
async def load(
|
||||
ctx: typer.Context,
|
||||
contexts: Annotated[
|
||||
list[str] | None,
|
||||
typer.Argument(
|
||||
help="Contexts to load (base, production, development, testing)."
|
||||
),
|
||||
list[Context] | None,
|
||||
typer.Argument(help="Contexts to load."),
|
||||
] = None,
|
||||
strategy: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
"--strategy", "-s", help="Load strategy: merge, insert, skip_existing."
|
||||
),
|
||||
] = "merge",
|
||||
LoadStrategy,
|
||||
typer.Option("--strategy", "-s", help="Load strategy."),
|
||||
] = LoadStrategy.MERGE,
|
||||
dry_run: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
@@ -79,19 +69,10 @@ async def load(
|
||||
] = False,
|
||||
) -> None:
|
||||
"""Load fixtures into the database."""
|
||||
config = _get_config(ctx)
|
||||
registry = config.get_fixtures_registry()
|
||||
get_db_context = config.get_db_context()
|
||||
registry = get_fixtures_registry()
|
||||
db_context = get_db_context()
|
||||
|
||||
context_list = contexts if contexts else [Context.BASE]
|
||||
|
||||
try:
|
||||
load_strategy = LoadStrategy(strategy)
|
||||
except ValueError:
|
||||
typer.echo(
|
||||
f"Invalid strategy: {strategy}. Use: merge, insert, skip_existing", err=True
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
context_list = [c.value for c in contexts] if contexts else [Context.BASE]
|
||||
|
||||
ordered = registry.resolve_context_dependencies(*context_list)
|
||||
|
||||
@@ -99,7 +80,7 @@ async def load(
|
||||
print("No fixtures to load for the specified context(s).")
|
||||
return
|
||||
|
||||
print(f"\nFixtures to load ({load_strategy.value} strategy):")
|
||||
print(f"\nFixtures to load ({strategy.value} strategy):")
|
||||
for name in ordered:
|
||||
fixture = registry.get(name)
|
||||
instances = list(fixture.func())
|
||||
@@ -110,9 +91,9 @@ async def load(
|
||||
print("\n[Dry run - no changes made]")
|
||||
return
|
||||
|
||||
async with get_db_context() as session:
|
||||
async with db_context() as session:
|
||||
result = await load_fixtures_by_context(
|
||||
session, registry, *context_list, strategy=load_strategy
|
||||
session, registry, *context_list, strategy=strategy
|
||||
)
|
||||
|
||||
total = sum(len(items) for items in result.values())
|
||||
|
||||
@@ -1,52 +1,28 @@
|
||||
"""CLI configuration."""
|
||||
"""CLI configuration and dynamic imports."""
|
||||
|
||||
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)
|
||||
from .pyproject import find_pyproject, load_pyproject
|
||||
|
||||
|
||||
def _import_from_string(import_path: str):
|
||||
"""Import an object from a string path like 'module.submodule:attribute'."""
|
||||
def _ensure_project_in_path():
|
||||
"""Add project root to sys.path if not installed in editable mode."""
|
||||
pyproject = find_pyproject()
|
||||
if pyproject:
|
||||
project_root = str(pyproject.parent)
|
||||
if project_root not in sys.path:
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
|
||||
def import_from_string(import_path: str):
|
||||
"""Import an object from a string path like 'module.submodule:attribute'.
|
||||
|
||||
Raises:
|
||||
typer.BadParameter: If the import path is invalid or import fails.
|
||||
"""
|
||||
if ":" not in import_path:
|
||||
raise typer.BadParameter(
|
||||
f"Invalid import path '{import_path}'. Expected format: 'module:attribute'"
|
||||
@@ -54,10 +30,7 @@ def _import_from_string(import_path: str):
|
||||
|
||||
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)
|
||||
_ensure_project_in_path()
|
||||
|
||||
try:
|
||||
module = importlib.import_module(module_path)
|
||||
@@ -72,21 +45,63 @@ def _import_from_string(import_path: str):
|
||||
return getattr(module, attr_name)
|
||||
|
||||
|
||||
def load_config() -> CliConfig:
|
||||
"""Load CLI configuration from pyproject.toml."""
|
||||
pyproject_path = Path.cwd() / "pyproject.toml"
|
||||
def get_config_value(key: str, required: bool = False):
|
||||
"""Get a configuration value from pyproject.toml.
|
||||
|
||||
if not pyproject_path.exists():
|
||||
return CliConfig()
|
||||
Args:
|
||||
key: The configuration key in [tool.fastapi-toolsets].
|
||||
required: If True, raises an error when the key is missing.
|
||||
|
||||
try:
|
||||
with open(pyproject_path, "rb") as f:
|
||||
data = tomllib.load(f)
|
||||
Returns:
|
||||
The configuration value, or None if not found and not required.
|
||||
|
||||
tool_config = data.get("tool", {}).get("fastapi-toolsets", {})
|
||||
return CliConfig(
|
||||
fixtures=tool_config.get("fixtures"),
|
||||
db_context=tool_config.get("db_context"),
|
||||
Raises:
|
||||
typer.BadParameter: If required=True and the key is missing.
|
||||
"""
|
||||
config = load_pyproject()
|
||||
value = config.get(key)
|
||||
|
||||
if required and value is None:
|
||||
raise typer.BadParameter(
|
||||
f"No '{key}' configured. "
|
||||
f"Add '{key}' to [tool.fastapi-toolsets] in pyproject.toml."
|
||||
)
|
||||
except Exception:
|
||||
return CliConfig()
|
||||
|
||||
return value
|
||||
|
||||
|
||||
def get_fixtures_registry():
|
||||
"""Import and return the fixtures registry from config."""
|
||||
from ..fixtures import FixtureRegistry
|
||||
|
||||
import_path = get_config_value("fixtures", required=True)
|
||||
registry = import_from_string(import_path)
|
||||
|
||||
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():
|
||||
"""Import and return the db_context function from config."""
|
||||
import_path = get_config_value("db_context", required=True)
|
||||
return import_from_string(import_path)
|
||||
|
||||
|
||||
def get_custom_cli() -> typer.Typer | None:
|
||||
"""Import and return the custom CLI Typer instance from config."""
|
||||
import_path = get_config_value("custom_cli")
|
||||
if not import_path:
|
||||
return None
|
||||
|
||||
custom = import_from_string(import_path)
|
||||
|
||||
if not isinstance(custom, typer.Typer):
|
||||
raise typer.BadParameter(
|
||||
f"'custom_cli' must be a Typer instance, got {type(custom).__name__}"
|
||||
)
|
||||
|
||||
return custom
|
||||
|
||||
43
src/fastapi_toolsets/cli/pyproject.py
Normal file
43
src/fastapi_toolsets/cli/pyproject.py
Normal file
@@ -0,0 +1,43 @@
|
||||
"""Pyproject.toml discovery and loading."""
|
||||
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
TOOL_NAME = "fastapi-toolsets"
|
||||
|
||||
|
||||
def find_pyproject(start_path: Path | None = None) -> Path | None:
|
||||
"""Find pyproject.toml by walking up the directory tree.
|
||||
|
||||
Similar to how pytest, black, and ruff discover their config files.
|
||||
"""
|
||||
path = (start_path or Path.cwd()).resolve()
|
||||
|
||||
for directory in [path, *path.parents]:
|
||||
pyproject = directory / "pyproject.toml"
|
||||
if pyproject.is_file():
|
||||
return pyproject
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def load_pyproject(path: Path | None = None) -> dict:
|
||||
"""Load tool configuration from pyproject.toml.
|
||||
|
||||
Args:
|
||||
path: Explicit path to pyproject.toml. If None, searches up from cwd.
|
||||
|
||||
Returns:
|
||||
The [tool.fastapi-toolsets] section as a dict, or empty dict if not found.
|
||||
"""
|
||||
pyproject_path = path or find_pyproject()
|
||||
|
||||
if not pyproject_path:
|
||||
return {}
|
||||
|
||||
try:
|
||||
with open(pyproject_path, "rb") as f:
|
||||
data = tomllib.load(f)
|
||||
return data.get("tool", {}).get(TOOL_NAME, {})
|
||||
except (OSError, tomllib.TOMLDecodeError):
|
||||
return {}
|
||||
Reference in New Issue
Block a user