Compare commits

...
3 Commits
4 changed files with 25 additions and 4 deletions
+14
View File
@@ -68,6 +68,20 @@ manager fixtures --help
╰──────────────────────────────────────────────────────────────────────────────────╯ ╰──────────────────────────────────────────────────────────────────────────────────╯
``` ```
### `fixtures load`
```bash
manager fixtures load [CONTEXTS]... [--strategy merge|insert|skip_existing] [--dry-run]
```
`CONTEXTS` defaults to `Context.BASE` when omitted, and can also be set via the `FIXTURES_CONTEXT` environment variable, handy for CI/deploy scripts that shouldn't need an explicit argument per environment:
```bash
FIXTURES_CONTEXT=testing manager fixtures load
```
An explicit CLI argument always takes precedence over the environment variable.
## Custom CLI ## Custom CLI
You can extend the CLI by providing your own Typer app. The `manager` entry point will merge your app's commands with the built-in ones: You can extend the CLI by providing your own Typer app. The `manager` entry point will merge your app's commands with the built-in ones:
@@ -57,7 +57,7 @@ async def load(
ctx: typer.Context, ctx: typer.Context,
contexts: Annotated[ contexts: Annotated[
list[str] | None, list[str] | None,
typer.Argument(help="Contexts to load."), typer.Argument(help="Contexts to load.", envvar="FIXTURES_CONTEXT"),
] = None, ] = None,
strategy: Annotated[ strategy: Annotated[
LoadStrategy, LoadStrategy,
@@ -1,7 +1,13 @@
"""Fixture system for seeding databases with dependency resolution.""" """Fixture system for seeding databases with dependency resolution."""
from typing import TYPE_CHECKING
from .enum import Context, LoadStrategy from .enum import Context, LoadStrategy
if TYPE_CHECKING:
from .registry import FixtureRegistry
from .utils import load_fixtures, load_fixtures_by_context
__all__ = [ __all__ = [
"Context", "Context",
"FixtureRegistry", "FixtureRegistry",
+4 -3
View File
@@ -139,12 +139,13 @@ async def _batch_merge(
) -> None: ) -> None:
"""UPSERT: insert new rows, update existing ones with the provided values.""" """UPSERT: insert new rows, update existing ones with the provided values."""
for cls, group_dicts, _ in _grouped_table_dicts(model_cls, instances): for cls, group_dicts, _ in _grouped_table_dicts(model_cls, instances):
pk_names = [col.name for col in cls.__table__.primary_key] table = cast(Table, cls.__table__)
pk_names = [col.name for col in table.primary_key]
pk_names_set = set(pk_names) pk_names_set = set(pk_names)
own_col_keys = {col.key for col in cls.__table__.columns} own_col_keys = {col.key for col in table.columns}
non_pk_cols = [k for k in own_col_keys if k not in pk_names_set] non_pk_cols = [k for k in own_col_keys if k not in pk_names_set]
stmt = pg_insert(cls).values(group_dicts) stmt = pg_insert(table).values(group_dicts)
inserted_keys = set(group_dicts[0]) inserted_keys = set(group_dicts[0])
update_cols = [col for col in non_pk_cols if col in inserted_keys] update_cols = [col for col in non_pk_cols if col in inserted_keys]