diff --git a/AGENTS.md b/AGENTS.md index 458395a..4bc5916 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -9,6 +9,7 @@ This file is for anyone changing this repository, human or agent. It states the | `make install` | Installs every dependency group. | | `make hooks` | Installs the pre-commit hooks so `make check` runs before each commit. | | `make check` | Runs every lint gate and the unit tests. Green here means green in CI. | +| `make seed` | Loads `seed/fixtures.py`. A second run changes nothing. | | `make test-integration` | Runs the Postgres tests in a disposable container. Needs Docker. | | `make test-cov` | Runs all tests with the coverage floor from `pyproject.toml`. | | `make openapi-snapshot` | Regenerates `tests/contract/openapi.json` after an intended API change. | diff --git a/Dockerfile.migration b/Dockerfile.migration index 11ff411..1a007c3 100644 --- a/Dockerfile.migration +++ b/Dockerfile.migration @@ -32,6 +32,7 @@ COPY --from=builder /app/.venv /app/.venv COPY alembic.ini /app/ COPY alembic /app/alembic +COPY seed /app/seed ENV PATH="/app/.venv/bin:$PATH" ENV PYTHONUNBUFFERED=1 diff --git a/Makefile b/Makefile index 774bbb1..f8477c6 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help install upgrade hooks lint lint-ruff lint-format lint-ty lint-imports lint-typos format check run migrate docker-up docker-down docker-down-destroy test test-unit test-integration test-cov openapi-snapshot +.PHONY: help install upgrade hooks lint lint-ruff lint-format lint-ty lint-imports lint-typos format check run migrate seed docker-up docker-down docker-down-destroy test test-unit test-integration test-cov openapi-snapshot help: @echo "Available targets:" @@ -10,6 +10,7 @@ help: @echo " make lint-imports - Run import-linter architecture checks" @echo " make format - Auto-fix ruff findings and format the code" @echo " make migrate - Apply Alembic migrations to the configured database" + @echo " make seed - Load fixture data; safe to run repeatedly" @echo " make run - Apply migrations, then start FastAPI dev server with hot reload" @echo " make docker-up - Start full project locally" @echo " make docker-down - Stop full project locally" @@ -34,13 +35,13 @@ check: lint test-unit lint: lint-ruff lint-format lint-ty lint-imports lint-typos lint-ruff: - uv run --group lint --all-packages ruff check app tests alembic scenario-tests + uv run --group lint --all-packages ruff check app tests alembic seed scenario-tests lint-format: - uv run --group lint --all-packages ruff format --check app tests alembic scenario-tests + uv run --group lint --all-packages ruff format --check app tests alembic seed scenario-tests lint-ty: - uv run --group lint --group migration --group test --all-packages ty check --error-on-warning app tests alembic scenario-tests + uv run --group lint --group migration --group test --all-packages ty check --error-on-warning app tests alembic seed scenario-tests lint-imports: uv run --group lint --all-packages lint-imports --config .importlinter @@ -49,8 +50,8 @@ lint-typos: uvx typos format: - uv run --group lint --all-packages ruff check --fix app tests alembic scenario-tests - uv run --group lint --all-packages ruff format app tests alembic scenario-tests + uv run --group lint --all-packages ruff check --fix app tests alembic seed scenario-tests + uv run --group lint --all-packages ruff format app tests alembic seed scenario-tests run: @if [ ! -f .env ]; then \ @@ -64,6 +65,9 @@ run: migrate: uv run --group migration --env-file .env alembic -c alembic.ini upgrade head +seed: + uv run --group migration --env-file .env python -m seed + docker-up: docker compose up --build diff --git a/docker-compose.yml b/docker-compose.yml index 24dcc3a..0a9cfa9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -72,6 +72,17 @@ services: condition: service_healthy restart: "no" + seed: + <<: *migration-image + profiles: ["seed"] + command: ["python", "-m", "seed"] + environment: + DATABASE_URL: postgresql+psycopg://postgres:${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required}@db:5432/fastapi_db + depends_on: + migrate: + condition: service_completed_successfully + restart: "no" + app: <<: *app-image ports: diff --git a/docs/local-development.md b/docs/local-development.md index a40336a..8eceff6 100644 --- a/docs/local-development.md +++ b/docs/local-development.md @@ -47,6 +47,10 @@ PostgreSQL data survives a normal `docker compose down`. Removing volumes resets docker compose down -v ``` +## Seed data + +Seeding is a separate command and never runs on application startup. `make seed` loads the rows in `seed/fixtures.py` into the database from `.env`. In the Compose stack, run `docker compose run --rm seed`. Every fixture row sets its primary key. A row whose primary key already exists is left as it is, so running the command again changes nothing. For an integer identity key, the command then moves the sequence past the highest stored ID if it is not already past it, so the next insert gets a new one. This assumes the ascending, step-1 sequence that SQLAlchemy's `Identity()` creates. + ## Isolated stacks `COMPOSE_PROJECT_NAME` separates container names, networks, volumes, and default image tags. Parallel stacks must also use distinct host ports and public URLs: diff --git a/seed/__init__.py b/seed/__init__.py new file mode 100644 index 0000000..5685ebb --- /dev/null +++ b/seed/__init__.py @@ -0,0 +1,91 @@ +from collections.abc import Mapping, Sequence +from itertools import groupby + +from sqlalchemy import ( + Connection, + MetaData, + Table, + cast, + column, + func, + select, + table, + text, +) +from sqlalchemy.dialects.postgresql import REGCLASS, insert + +type Row = Mapping[str, object] +type Fixtures = Mapping[str, Sequence[Row]] + +_PG_CLASS = table("pg_class", column("oid"), column("relname"), column("relnamespace")) +_PG_NAMESPACE = table("pg_namespace", column("oid"), column("nspname")) + + +def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: + metadata = MetaData() + metadata.reflect(bind=connection, only=list(fixtures)) + inserted: dict[str, int] = {} + for seeded_table in metadata.sorted_tables: + rows = fixtures.get(seeded_table.name) + if seeded_table.schema is not None or not rows: + continue + _require_primary_keys(seeded_table, rows) + _lock_against_writers(connection, seeded_table) + inserted[seeded_table.name] = _insert_rows(connection, seeded_table, rows) + _restart_identity_sequences(connection, seeded_table) + return inserted + + +def _require_primary_keys(seeded_table: Table, rows: Sequence[Row]) -> None: + key = [column.name for column in seeded_table.primary_key.columns] + if not key or any(row.get(name) is None for row in rows for name in key): + msg = f"{seeded_table.name} fixture rows must set the primary key {key}" + raise ValueError(msg) + + +def _lock_against_writers(connection: Connection, seeded_table: Table) -> None: + table_name = connection.dialect.identifier_preparer.format_table(seeded_table) + connection.execute(text(f"LOCK TABLE {table_name} IN SHARE ROW EXCLUSIVE MODE")) + + +def _insert_rows( + connection: Connection, seeded_table: Table, rows: Sequence[Row] +) -> int: + key = list(seeded_table.primary_key.columns) + statement = ( + insert(seeded_table).on_conflict_do_nothing(index_elements=key).returning(*key) + ) + return sum( + len(connection.execute(statement, list(batch)).all()) + for _, batch in groupby(rows, key=frozenset) + ) + + +def _restart_identity_sequences(connection: Connection, seeded_table: Table) -> None: + table_name = connection.dialect.identifier_preparer.format_table(seeded_table) + for key_column in seeded_table.primary_key.columns: + sequence = connection.scalar( + select(func.pg_get_serial_sequence(table_name, key_column.name)) + ) + if sequence is None: + continue + highest = connection.execute(select(func.max(key_column))).scalar_one() + if highest >= _next_value(connection, sequence): + connection.execute( + text(f"ALTER SEQUENCE {sequence} RESTART WITH {int(highest) + 1}") + ) + + +def _next_value(connection: Connection, sequence: str) -> int: + schema, name = connection.execute( + select(_PG_NAMESPACE.c.nspname, _PG_CLASS.c.relname) + .join_from( + _PG_CLASS, _PG_NAMESPACE, _PG_CLASS.c.relnamespace == _PG_NAMESPACE.c.oid + ) + .where(_PG_CLASS.c.oid == cast(sequence, REGCLASS)) + ).one() + state = table(name, column("last_value"), column("is_called"), schema=schema) + last_value, is_called = connection.execute( + select(state.c.last_value, state.c.is_called) + ).one() + return last_value + 1 if is_called else last_value diff --git a/seed/__main__.py b/seed/__main__.py new file mode 100644 index 0000000..3a23cec --- /dev/null +++ b/seed/__main__.py @@ -0,0 +1,29 @@ +import logging +import os + +from sqlalchemy import create_engine, pool + +from seed import seed +from seed.fixtures import FIXTURES + +logger = logging.getLogger("seed") + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(message)s") + database_url = os.environ.get("DATABASE_URL") + if not database_url: + msg = "DATABASE_URL must be set to seed the database" + raise RuntimeError(msg) + engine = create_engine(database_url, poolclass=pool.NullPool) + try: + with engine.begin() as connection: + inserted = seed(connection, FIXTURES) + finally: + engine.dispose() + for table, count in inserted.items(): + logger.info("%s: %d new rows", table, count) + + +if __name__ == "__main__": + main() diff --git a/seed/fixtures.py b/seed/fixtures.py new file mode 100644 index 0000000..03d510b --- /dev/null +++ b/seed/fixtures.py @@ -0,0 +1,18 @@ +from uuid import UUID + +from seed import Fixtures + +FIXTURES: Fixtures = { + "entities": [ + { + "id": UUID("6f1c2a52-8d3b-4c1e-9a4f-0b7e5d2c1a01"), + "name": "Sample entity", + "description": "Reference data loaded by make seed.", + }, + { + "id": UUID("6f1c2a52-8d3b-4c1e-9a4f-0b7e5d2c1a02"), + "name": "Second sample entity", + "description": None, + }, + ], +} diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py new file mode 100644 index 0000000..bb148cc --- /dev/null +++ b/tests/integration/test_seed.py @@ -0,0 +1,220 @@ +from collections.abc import Iterator + +import pytest +from asgi_lifespan import LifespanManager +from sqlalchemy import Connection, create_engine, delete, func, select, text +from sqlalchemy.exc import OperationalError +from sqlalchemy.ext.asyncio import AsyncEngine + +from app.auth import OIDCOpenAPIFastAPI +from app.model import Entity +from seed import seed +from seed.fixtures import FIXTURES + +pytestmark = pytest.mark.integration + +IDENTITY_TABLE = ( + "CREATE TABLE {name} (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" +) + + +@pytest.fixture +def connection(test_postgres_url: str) -> Iterator[Connection]: + engine = create_engine(test_postgres_url) + try: + with engine.connect() as connection: + yield connection + connection.rollback() + finally: + engine.dispose() + + +def _entity_rows(connection: Connection) -> list[tuple[object, ...]]: + statement = select(Entity.__table__).order_by(Entity.id) + return [tuple(row) for row in connection.execute(statement)] + + +def test_seeding_twice_changes_nothing(connection: Connection) -> None: + connection.execute(delete(Entity)) + + first = seed(connection, FIXTURES) + rows_after_first = _entity_rows(connection) + second = seed(connection, FIXTURES) + + assert first == {"entities": len(FIXTURES["entities"])} + assert second == {"entities": 0} + assert _entity_rows(connection) == rows_after_first + + +def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> None: + connection.execute(text(IDENTITY_TABLE.format(name="seed_probe"))) + + seed(connection, {"seed_probe": [{"id": 1, "name": "a"}, {"id": 7, "name": "b"}]}) + seed(connection, {"seed_probe": [{"id": 1, "name": "a"}, {"id": 7, "name": "b"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 8 + + +def test_rerun_keeps_a_sequence_that_is_already_ahead(connection: Connection) -> None: + connection.execute(text(IDENTITY_TABLE.format(name="seed_probe"))) + seed(connection, {"seed_probe": [{"id": 1, "name": "a"}]}) + connection.execute( + text("INSERT INTO seed_probe (name) SELECT 'x' FROM generate_series(1, 4)") + ) + connection.execute(text("DELETE FROM seed_probe WHERE id > 1")) + + seed(connection, {"seed_probe": [{"id": 1, "name": "a"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 6 + + +def test_rows_without_a_primary_key_are_rejected(connection: Connection) -> None: + connection.execute(text(IDENTITY_TABLE.format(name="seed_probe"))) + + with pytest.raises(ValueError, match=r"must set the primary key \['id'\]"): + seed(connection, {"seed_probe": [{"name": "a"}]}) + + +def test_rows_with_different_columns_keep_their_own_values( + connection: Connection, +) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer PRIMARY KEY, name text, note text DEFAULT 'none')" + ) + ) + + seed( + connection, + { + "seed_probe": [ + {"id": 1, "name": "a"}, + {"id": 2, "name": "b", "note": "set"}, + ] + }, + ) + notes = connection.execute( + text("SELECT name, note FROM seed_probe ORDER BY id") + ).all() + + assert [tuple(row) for row in notes] == [("a", "none"), ("b", "set")] + + +def test_rows_are_inserted_in_fixture_order(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer PRIMARY KEY, parent integer REFERENCES seed_probe (id), " + "note text)" + ) + ) + + inserted = seed( + connection, + { + "seed_probe": [ + {"id": 1, "parent": None}, + {"id": 2, "parent": 1, "note": "x"}, + {"id": 3, "parent": 2}, + ] + }, + ) + + assert inserted == {"seed_probe": 3} + + +def test_tables_in_other_schemas_are_not_seeded(connection: Connection) -> None: + connection.execute(text("CREATE SCHEMA seed_other")) + connection.execute( + text("CREATE TABLE seed_other.seed_probe (id integer PRIMARY KEY, name text)") + ) + connection.execute( + text("CREATE TABLE seed_probe (id integer PRIMARY KEY, name text)") + ) + connection.execute( + text( + "CREATE TABLE seed_child (id integer PRIMARY KEY, " + "other_id integer REFERENCES seed_other.seed_probe (id))" + ) + ) + + inserted = seed( + connection, + { + "seed_probe": [{"id": 1, "name": "public"}], + "seed_child": [{"id": 1, "other_id": None}], + }, + ) + other_rows = connection.scalar(text("SELECT count(*) FROM seed_other.seed_probe")) + + assert inserted == {"seed_probe": 1, "seed_child": 1} + assert other_rows == 0 + + +def test_rolled_back_seed_leaves_the_sequence_alone(test_postgres_url: str) -> None: + engine = create_engine(test_postgres_url) + try: + with engine.begin() as setup: + setup.execute(text(IDENTITY_TABLE.format(name="seed_rollback_probe"))) + with engine.connect() as seeding: + seed(seeding, {"seed_rollback_probe": [{"id": 500, "name": "seeded"}]}) + seeding.rollback() + with engine.begin() as writer: + new_id = writer.scalar( + text( + "INSERT INTO seed_rollback_probe (name) VALUES ('writer') " + "RETURNING id" + ) + ) + finally: + with engine.begin() as cleanup: + cleanup.execute(text("DROP TABLE IF EXISTS seed_rollback_probe")) + engine.dispose() + + assert new_id == 1 + + +def test_concurrent_insert_waits_until_seeding_commits(test_postgres_url: str) -> None: + engine = create_engine(test_postgres_url) + insert_row = text("INSERT INTO seed_lock_probe (id, name) VALUES (6, 'writer')") + try: + with engine.begin() as setup: + setup.execute( + text("CREATE TABLE seed_lock_probe (id integer PRIMARY KEY, name text)") + ) + with engine.connect() as seeding, engine.connect() as writer: + seed(seeding, {"seed_lock_probe": [{"id": 5, "name": "seeded"}]}) + writer.execute(text("SET lock_timeout = '200ms'")) + readable = writer.scalar(text("SELECT count(*) FROM seed_lock_probe")) + with pytest.raises(OperationalError, match="lock timeout"): + writer.execute(insert_row) + writer.rollback() + seeding.commit() + writer.execute(insert_row) + writer.commit() + rows = writer.scalar(text("SELECT count(*) FROM seed_lock_probe")) + finally: + with engine.begin() as cleanup: + cleanup.execute(text("DROP TABLE IF EXISTS seed_lock_probe")) + engine.dispose() + + assert readable == 0 + assert rows == 2 + + +async def test_application_startup_does_not_seed( + integration_app: tuple[OIDCOpenAPIFastAPI, LifespanManager], + integration_engine: AsyncEngine, +) -> None: + async with integration_engine.connect() as connection: + count = await connection.scalar(select(func.count()).select_from(Entity)) + + assert count == 0