Rework fixtures module (#333)

* feat: fixture refresh DB-generated values onto returned instances

* refactor: replace utils.get_obj_by_attr/get_field_by_attr with registry.obj/field lookups

* refactor: log fixture command output instead of print

* chore: clean up fixture module
This commit is contained in:
d3vyce
2026-07-01 19:24:00 +02:00
committed by GitHub
parent 70e0b3b9d5
commit fe2c0f3eff
10 changed files with 359 additions and 295 deletions
+15 -14
View File
@@ -7,6 +7,7 @@ from rich.console import Console
from rich.table import Table
from ...fixtures import Context, LoadStrategy, load_fixtures_by_context
from ...logger import get_logger
from ..config import get_db_context, get_fixtures_registry
from ..utils import async_command
@@ -16,6 +17,7 @@ fixture_cli = typer.Typer(
no_args_is_help=True,
)
console = Console()
logger = get_logger()
@fixture_cli.command("list")
@@ -32,10 +34,10 @@ def list_fixtures(
) -> None:
"""List all registered fixtures."""
registry = get_fixtures_registry()
fixtures = registry.get_by_context(context.value) if context else registry.get_all()
fixtures = registry.get_by_context(context) if context else registry.get_all()
if not fixtures:
print("No fixtures found.")
logger.info("No fixtures found.")
return
table = Table("Name", "Contexts", "Dependencies")
@@ -46,7 +48,7 @@ def list_fixtures(
table.add_row(fixture.name, contexts, deps)
console.print(table)
print(f"\nTotal: {len(fixtures)} fixture(s)")
logger.info("Total: %d fixture(s)", len(fixtures))
@fixture_cli.command("load")
@@ -72,23 +74,22 @@ async def load(
registry = get_fixtures_registry()
db_context = get_db_context()
context_list = list(contexts) if contexts else [Context.BASE]
context_list = contexts or [Context.BASE]
ordered = registry.resolve_context_dependencies(*context_list)
if not ordered:
print("No fixtures to load for the specified context(s).")
logger.info("No fixtures to load for the specified context(s).")
return
print(f"\nFixtures to 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 "?"
print(f" - {name}: {len(instances)} {model_name}(s)")
if dry_run:
print("\n[Dry run - no changes made]")
logger.info("Fixtures to load (%s strategy):", strategy.value)
for name in ordered:
variants = registry.get_load_variants(name, *context_list)
instances = [inst for v in variants for inst in v.func()]
model_name = type(instances[0]).__name__ if instances else "?"
logger.info(" - %s: %d %s(s)", name, len(instances), model_name)
logger.info("[Dry run - no changes made]")
return
async with db_context() as session:
@@ -97,4 +98,4 @@ async def load(
)
total = sum(len(items) for items in result.values())
print(f"\nLoaded {total} record(s) successfully.")
logger.info("Loaded %d record(s) successfully.", total)
+33 -22
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import importlib
import sys
from typing import TYPE_CHECKING, Any, Literal, overload
from typing import TYPE_CHECKING, Any, Literal, TypeVar, overload
import typer
@@ -13,6 +13,8 @@ from .pyproject import find_pyproject, load_pyproject
if TYPE_CHECKING:
from ..fixtures import FixtureRegistry
T = TypeVar("T")
def _ensure_project_in_path():
"""Add project root to sys.path if not installed in editable mode."""
@@ -88,19 +90,39 @@ def get_config_value(key: str, required: bool = False) -> Any | None:
return value
@overload
def _import_typed(
key: str, expected_type: type[T], *, required: Literal[True]
) -> T: ... # pragma: no cover
@overload
def _import_typed(
key: str, expected_type: type[T], *, required: bool
) -> T | None: ... # pragma: no cover
def _import_typed(key: str, expected_type: type[T], *, required: bool) -> T | None:
"""Import a config value by key and validate its type.
Raises:
typer.BadParameter: If required and missing, or if the imported
value isn't an instance of *expected_type*.
"""
import_path = get_config_value(key, required=required)
if not import_path:
return None
obj = import_from_string(import_path)
if not isinstance(obj, expected_type):
raise typer.BadParameter(
f"'{key}' must be a {expected_type.__name__} instance, got {type(obj).__name__}"
)
return obj
def get_fixtures_registry() -> FixtureRegistry:
"""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
return _import_typed("fixtures", FixtureRegistry, required=True)
def get_db_context() -> Any:
@@ -111,15 +133,4 @@ def get_db_context() -> Any:
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
return _import_typed("custom_cli", typer.Typer, required=False)