From b225fb350b4c3c16baf4ea63012588e1c8f5ae3e Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 22:29:05 +0530 Subject: [PATCH 01/13] Add idempotent seed-data command --- AGENTS.md | 1 + Dockerfile.migration | 1 + Makefile | 16 ++++--- docker-compose.yml | 11 +++++ docs/local-development.md | 4 ++ seed/__init__.py | 45 ++++++++++++++++++ seed/__main__.py | 29 ++++++++++++ seed/fixtures.py | 18 +++++++ tests/integration/test_seed.py | 87 ++++++++++++++++++++++++++++++++++ 9 files changed, 206 insertions(+), 6 deletions(-) create mode 100644 seed/__init__.py create mode 100644 seed/__main__.py create mode 100644 seed/fixtures.py create mode 100644 tests/integration/test_seed.py 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..8b23b33 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`. Rows that already exist are kept as they are, so running the command again changes nothing. After inserting explicit IDs, the command moves each serial or identity sequence past the highest ID, so the next insert gets a new one. + ## 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..617d36f --- /dev/null +++ b/seed/__init__.py @@ -0,0 +1,45 @@ +from collections.abc import Mapping, Sequence + +from sqlalchemy import Column, Connection, MetaData, Table, cast, func, select +from sqlalchemy.dialects.postgresql import REGCLASS, insert + +type Row = Mapping[str, object] +type Fixtures = Mapping[str, Sequence[Row]] + + +def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: + metadata = MetaData() + metadata.reflect(bind=connection, only=list(fixtures)) + inserted: dict[str, int] = {} + for table in metadata.sorted_tables: + rows = fixtures.get(table.name) + if not rows: + continue + statement = insert(table).on_conflict_do_nothing().returning(*table.c) + inserted[table.name] = len(connection.execute(statement, list(rows)).all()) + _advance_sequences(connection, table) + return inserted + + +def _advance_sequences(connection: Connection, table: Table) -> None: + table_name = connection.dialect.identifier_preparer.format_table(table) + for column in table.c: + sequence = connection.scalar( + select(func.pg_get_serial_sequence(table_name, column.name)) + ) + if sequence is not None: + _advance_sequence(connection, table, column, sequence) + + +def _advance_sequence( + connection: Connection, + table: Table, + column: Column[object], + sequence: str, +) -> None: + sequence_id = cast(sequence, REGCLASS) + highest = select(func.max(column)).select_from(table).scalar_subquery() + last_value = func.coalesce(func.pg_sequence_last_value(sequence_id), highest) + connection.execute( + select(func.setval(sequence_id, func.greatest(highest, 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..2e789b9 --- /dev/null +++ b/tests/integration/test_seed.py @@ -0,0 +1,87 @@ +from collections.abc import Iterator + +import pytest +from asgi_lifespan import LifespanManager +from sqlalchemy import Connection, create_engine, delete, func, select, text +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 + + +@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( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" + ) + ) + + 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_sequence_never_moves_backwards(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" + ) + ) + connection.execute( + text("INSERT INTO seed_probe (name) SELECT 'x' FROM generate_series(1, 20)") + ) + connection.execute(text("DELETE FROM seed_probe WHERE id > 2")) + + seed(connection, {"seed_probe": [{"id": 3, "name": "seeded"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 21 + + +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 From 93e74cad5a9d3bdf58272a0c70345dc308c63f92 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 22:35:24 +0530 Subject: [PATCH 02/13] Lock seeded tables while moving sequences --- seed/__init__.py | 8 +++++++- tests/integration/test_seed.py | 35 ++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/seed/__init__.py b/seed/__init__.py index 617d36f..dd39f19 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -1,6 +1,6 @@ from collections.abc import Mapping, Sequence -from sqlalchemy import Column, Connection, MetaData, Table, cast, func, select +from sqlalchemy import Column, Connection, MetaData, Table, cast, func, select, text from sqlalchemy.dialects.postgresql import REGCLASS, insert type Row = Mapping[str, object] @@ -15,12 +15,18 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: rows = fixtures.get(table.name) if not rows: continue + _lock_against_writers(connection, table) statement = insert(table).on_conflict_do_nothing().returning(*table.c) inserted[table.name] = len(connection.execute(statement, list(rows)).all()) _advance_sequences(connection, table) return inserted +def _lock_against_writers(connection: Connection, table: Table) -> None: + table_name = connection.dialect.identifier_preparer.format_table(table) + connection.execute(text(f"LOCK TABLE {table_name} IN SHARE ROW EXCLUSIVE MODE")) + + def _advance_sequences(connection: Connection, table: Table) -> None: table_name = connection.dialect.identifier_preparer.format_table(table) for column in table.c: diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index 2e789b9..7b45924 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -3,6 +3,7 @@ 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 @@ -77,6 +78,40 @@ def test_sequence_never_moves_backwards(connection: Connection) -> None: assert new_id == 21 +def test_concurrent_insert_waits_until_seeding_commits(test_postgres_url: str) -> None: + engine = create_engine(test_postgres_url) + try: + with engine.begin() as setup: + setup.execute( + text( + "CREATE TABLE seed_lock_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY 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'")) + with pytest.raises(OperationalError, match="lock timeout"): + writer.execute( + text("INSERT INTO seed_lock_probe (name) VALUES ('writer')") + ) + writer.rollback() + seeding.commit() + new_id = writer.scalar( + text( + "INSERT INTO seed_lock_probe (name) VALUES ('writer') RETURNING id" + ) + ) + writer.commit() + finally: + with engine.begin() as cleanup: + cleanup.execute(text("DROP TABLE IF EXISTS seed_lock_probe")) + engine.dispose() + + assert new_id == 6 + + async def test_application_startup_does_not_seed( integration_app: tuple[OIDCOpenAPIFastAPI, LifespanManager], integration_engine: AsyncEngine, From bf67368340eb5d0c3690b368b976ff9736d556fc Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 22:37:54 +0530 Subject: [PATCH 03/13] Seed explicit IDs into always-generated identity columns --- seed/__init__.py | 26 ++++++++++++++++++++++++++ tests/integration/test_seed.py | 14 ++++++++++++-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/seed/__init__.py b/seed/__init__.py index dd39f19..4216c6d 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -1,4 +1,5 @@ from collections.abc import Mapping, Sequence +from typing import Literal from sqlalchemy import Column, Connection, MetaData, Table, cast, func, select, text from sqlalchemy.dialects.postgresql import REGCLASS, insert @@ -16,8 +17,17 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: if not rows: continue _lock_against_writers(connection, table) + always_identities = [ + column.name + for column in table.c + if column.identity is not None + and column.identity.always + and any(column.name in row for row in rows) + ] + _set_identity_generation(connection, table, always_identities, "BY DEFAULT") statement = insert(table).on_conflict_do_nothing().returning(*table.c) inserted[table.name] = len(connection.execute(statement, list(rows)).all()) + _set_identity_generation(connection, table, always_identities, "ALWAYS") _advance_sequences(connection, table) return inserted @@ -27,6 +37,22 @@ def _lock_against_writers(connection: Connection, table: Table) -> None: connection.execute(text(f"LOCK TABLE {table_name} IN SHARE ROW EXCLUSIVE MODE")) +def _set_identity_generation( + connection: Connection, + table: Table, + columns: Sequence[str], + generation: Literal["ALWAYS", "BY DEFAULT"], +) -> None: + preparer = connection.dialect.identifier_preparer + for column in columns: + connection.execute( + text( + f"ALTER TABLE {preparer.format_table(table)} " + f"ALTER COLUMN {preparer.quote(column)} SET GENERATED {generation}" + ) + ) + + def _advance_sequences(connection: Connection, table: Table) -> None: table_name = connection.dialect.identifier_preparer.format_table(table) for column in table.c: diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index 7b45924..96279c8 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -42,11 +42,14 @@ def test_seeding_twice_changes_nothing(connection: Connection) -> None: assert _entity_rows(connection) == rows_after_first -def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> None: +@pytest.mark.parametrize("generation", ["ALWAYS", "BY DEFAULT"]) +def test_next_insert_after_explicit_ids_gets_a_new_id( + connection: Connection, generation: str +) -> None: connection.execute( text( "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" + f"id integer GENERATED {generation} AS IDENTITY PRIMARY KEY, name text)" ) ) @@ -54,8 +57,15 @@ def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> new_id = connection.scalar( text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") ) + generation_after_seed = connection.scalar( + text( + "SELECT identity_generation FROM information_schema.columns " + "WHERE table_name = 'seed_probe' AND column_name = 'id'" + ) + ) assert new_id == 8 + assert generation_after_seed == generation def test_sequence_never_moves_backwards(connection: Connection) -> None: From 24b73fa533bc533371e98b763dd877a19103d0ca Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 22:47:51 +0530 Subject: [PATCH 04/13] Seed rows one at a time and respect sequence direction --- docs/local-development.md | 2 +- seed/__init__.py | 101 ++++++++++++++++++++------------- tests/integration/test_seed.py | 72 +++++++++++++++++++---- 3 files changed, 123 insertions(+), 52 deletions(-) diff --git a/docs/local-development.md b/docs/local-development.md index 8b23b33..7be2059 100644 --- a/docs/local-development.md +++ b/docs/local-development.md @@ -49,7 +49,7 @@ 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`. Rows that already exist are kept as they are, so running the command again changes nothing. After inserting explicit IDs, the command moves each serial or identity sequence past the highest ID, so the next insert gets a new one. +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`. Rows that already exist are kept as they are, so running the command again changes nothing. Explicit IDs need a serial or `GENERATED BY DEFAULT` identity column. After inserting them, the command moves the column's sequence past every stored ID, so the next insert gets a new one. ## Isolated stacks diff --git a/seed/__init__.py b/seed/__init__.py index 4216c6d..6bbec81 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -1,77 +1,96 @@ from collections.abc import Mapping, Sequence -from typing import Literal -from sqlalchemy import Column, Connection, MetaData, Table, cast, func, select, text +from sqlalchemy import ( + Column, + Connection, + MetaData, + Table, + cast, + func, + select, + table, + text, +) from sqlalchemy.dialects.postgresql import REGCLASS, insert type Row = Mapping[str, object] type Fixtures = Mapping[str, Sequence[Row]] +_PG_SEQUENCE = table("pg_sequence", Column("seqrelid"), Column("seqincrement")) + def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: metadata = MetaData() metadata.reflect(bind=connection, only=list(fixtures)) inserted: dict[str, int] = {} - for table in metadata.sorted_tables: - rows = fixtures.get(table.name) + for seeded_table in metadata.sorted_tables: + rows = fixtures.get(seeded_table.name) if not rows: continue - _lock_against_writers(connection, table) - always_identities = [ - column.name - for column in table.c - if column.identity is not None - and column.identity.always - and any(column.name in row for row in rows) - ] - _set_identity_generation(connection, table, always_identities, "BY DEFAULT") - statement = insert(table).on_conflict_do_nothing().returning(*table.c) - inserted[table.name] = len(connection.execute(statement, list(rows)).all()) - _set_identity_generation(connection, table, always_identities, "ALWAYS") - _advance_sequences(connection, table) + _reject_always_identity_values(seeded_table, rows) + _lock_against_writers(connection, seeded_table) + inserted[seeded_table.name] = sum( + _insert_row(connection, seeded_table, row) for row in rows + ) + _advance_sequences(connection, seeded_table) return inserted -def _lock_against_writers(connection: Connection, table: Table) -> None: - table_name = connection.dialect.identifier_preparer.format_table(table) +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 _set_identity_generation( - connection: Connection, - table: Table, - columns: Sequence[str], - generation: Literal["ALWAYS", "BY DEFAULT"], -) -> None: - preparer = connection.dialect.identifier_preparer - for column in columns: - connection.execute( - text( - f"ALTER TABLE {preparer.format_table(table)} " - f"ALTER COLUMN {preparer.quote(column)} SET GENERATED {generation}" +def _reject_always_identity_values(seeded_table: Table, rows: Sequence[Row]) -> None: + for column in seeded_table.c: + identity = column.identity + if identity is None or not identity.always: + continue + if any(column.name in row for row in rows): + msg = ( + f"{seeded_table.name}.{column.name} is GENERATED ALWAYS, so fixtures " + "cannot set it; use GENERATED BY DEFAULT for seeded IDs" ) - ) + raise ValueError(msg) -def _advance_sequences(connection: Connection, table: Table) -> None: - table_name = connection.dialect.identifier_preparer.format_table(table) - for column in table.c: +def _insert_row(connection: Connection, seeded_table: Table, row: Row) -> int: + statement = ( + insert(seeded_table) + .values(row) + .on_conflict_do_nothing() + .returning(*seeded_table.c) + ) + return len(connection.execute(statement).all()) + + +def _advance_sequences(connection: Connection, seeded_table: Table) -> None: + table_name = connection.dialect.identifier_preparer.format_table(seeded_table) + for column in seeded_table.c: sequence = connection.scalar( select(func.pg_get_serial_sequence(table_name, column.name)) ) if sequence is not None: - _advance_sequence(connection, table, column, sequence) + _advance_sequence(connection, seeded_table, column, sequence) def _advance_sequence( connection: Connection, - table: Table, + seeded_table: Table, column: Column[object], sequence: str, ) -> None: sequence_id = cast(sequence, REGCLASS) - highest = select(func.max(column)).select_from(table).scalar_subquery() - last_value = func.coalesce(func.pg_sequence_last_value(sequence_id), highest) - connection.execute( - select(func.setval(sequence_id, func.greatest(highest, last_value))) + increment = connection.scalar( + select(_PG_SEQUENCE.c.seqincrement).where( + _PG_SEQUENCE.c.seqrelid == sequence_id + ) + ) + furthest, further = ( + (func.max, func.greatest) + if increment is None or increment > 0 + else (func.min, func.least) ) + stored = select(furthest(column)).select_from(seeded_table).scalar_subquery() + last_value = func.coalesce(func.pg_sequence_last_value(sequence_id), stored) + connection.execute(select(func.setval(sequence_id, further(stored, last_value)))) diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index 96279c8..a59be44 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -42,14 +42,11 @@ def test_seeding_twice_changes_nothing(connection: Connection) -> None: assert _entity_rows(connection) == rows_after_first -@pytest.mark.parametrize("generation", ["ALWAYS", "BY DEFAULT"]) -def test_next_insert_after_explicit_ids_gets_a_new_id( - connection: Connection, generation: str -) -> None: +def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> None: connection.execute( text( "CREATE TABLE seed_probe (" - f"id integer GENERATED {generation} AS IDENTITY PRIMARY KEY, name text)" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" ) ) @@ -57,15 +54,68 @@ def test_next_insert_after_explicit_ids_gets_a_new_id( new_id = connection.scalar( text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") ) - generation_after_seed = connection.scalar( + + assert new_id == 8 + + +def test_explicit_ids_for_always_generated_identity_are_rejected( + connection: Connection, +) -> None: + connection.execute( text( - "SELECT identity_generation FROM information_schema.columns " - "WHERE table_name = 'seed_probe' AND column_name = 'id'" + "CREATE TABLE seed_probe (" + "id integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY, name text)" ) ) - assert new_id == 8 - assert generation_after_seed == generation + with pytest.raises(ValueError, match=r"seed_probe\.id is GENERATED ALWAYS"): + seed(connection, {"seed_probe": [{"id": 1, "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_descending_sequence_moves_past_the_lowest_id(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY " + "(INCREMENT BY -1 START WITH -1 MAXVALUE -1) PRIMARY KEY, name text)" + ) + ) + connection.execute( + text("INSERT INTO seed_probe (name) SELECT 'x' FROM generate_series(1, 2)") + ) + + seed(connection, {"seed_probe": [{"id": -5, "name": "seeded"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == -6 def test_sequence_never_moves_backwards(connection: Connection) -> None: @@ -102,6 +152,7 @@ def test_concurrent_insert_waits_until_seeding_commits(test_postgres_url: str) - 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( text("INSERT INTO seed_lock_probe (name) VALUES ('writer')") @@ -119,6 +170,7 @@ def test_concurrent_insert_waits_until_seeding_commits(test_postgres_url: str) - cleanup.execute(text("DROP TABLE IF EXISTS seed_lock_probe")) engine.dispose() + assert readable == 0 assert new_id == 6 From d95c5dcec324d64cc7cc535254a212d2c4eb01b5 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 22:56:41 +0530 Subject: [PATCH 05/13] Batch seed inserts and restart sequences transactionally --- seed/__init__.py | 56 +++++++++++++++++++++------------- tests/integration/test_seed.py | 46 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 22 deletions(-) diff --git a/seed/__init__.py b/seed/__init__.py index 6bbec81..e5795e4 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -16,6 +16,8 @@ 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")) _PG_SEQUENCE = table("pg_sequence", Column("seqrelid"), Column("seqincrement")) @@ -29,9 +31,7 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: continue _reject_always_identity_values(seeded_table, rows) _lock_against_writers(connection, seeded_table) - inserted[seeded_table.name] = sum( - _insert_row(connection, seeded_table, row) for row in rows - ) + inserted[seeded_table.name] = _insert_rows(connection, seeded_table, rows) _advance_sequences(connection, seeded_table) return inserted @@ -54,14 +54,16 @@ def _reject_always_identity_values(seeded_table: Table, rows: Sequence[Row]) -> raise ValueError(msg) -def _insert_row(connection: Connection, seeded_table: Table, row: Row) -> int: - statement = ( - insert(seeded_table) - .values(row) - .on_conflict_do_nothing() - .returning(*seeded_table.c) +def _insert_rows( + connection: Connection, seeded_table: Table, rows: Sequence[Row] +) -> int: + batches: dict[frozenset[str], list[Row]] = {} + for row in rows: + batches.setdefault(frozenset(row), []).append(row) + statement = insert(seeded_table).on_conflict_do_nothing().returning(*seeded_table.c) + return sum( + len(connection.execute(statement, batch).all()) for batch in batches.values() ) - return len(connection.execute(statement).all()) def _advance_sequences(connection: Connection, seeded_table: Table) -> None: @@ -80,17 +82,27 @@ def _advance_sequence( column: Column[object], sequence: str, ) -> None: - sequence_id = cast(sequence, REGCLASS) - increment = connection.scalar( - select(_PG_SEQUENCE.c.seqincrement).where( - _PG_SEQUENCE.c.seqrelid == sequence_id + schema, name, increment = connection.execute( + select( + _PG_NAMESPACE.c.nspname, + _PG_CLASS.c.relname, + _PG_SEQUENCE.c.seqincrement, ) + .join_from( + _PG_CLASS, _PG_NAMESPACE, _PG_CLASS.c.relnamespace == _PG_NAMESPACE.c.oid + ) + .join(_PG_SEQUENCE, _PG_SEQUENCE.c.seqrelid == _PG_CLASS.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() + next_value = last_value + increment if is_called else last_value + furthest = func.max if increment > 0 else func.min + stored = connection.scalar(select(furthest(column)).select_from(seeded_table)) + if stored is None or (next_value - stored) * increment > 0: + return + connection.execute( + text(f"ALTER SEQUENCE {sequence} RESTART WITH {int(stored + increment)}") ) - furthest, further = ( - (func.max, func.greatest) - if increment is None or increment > 0 - else (func.min, func.least) - ) - stored = select(furthest(column)).select_from(seeded_table).scalar_subquery() - last_value = func.coalesce(func.pg_sequence_last_value(sequence_id), stored) - connection.execute(select(func.setval(sequence_id, further(stored, last_value)))) diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index a59be44..ac111ba 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -118,6 +118,52 @@ def test_descending_sequence_moves_past_the_lowest_id(connection: Connection) -> assert new_id == -6 +def test_unused_sequence_keeps_its_configured_start(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY " + "(START WITH 100 MINVALUE 100) PRIMARY KEY, name text)" + ) + ) + + seed(connection, {"seed_probe": [{"id": 1, "name": "seeded"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 100 + + +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( + "CREATE TABLE seed_rollback_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " + "name text)" + ) + ) + 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_sequence_never_moves_backwards(connection: Connection) -> None: connection.execute( text( From 85de8d0f0f815c9ffaf3a9cac13d524831f51743 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 23:07:10 +0530 Subject: [PATCH 06/13] Keep fixture order and sequence state on seed reruns --- seed/__init__.py | 60 +++++++++++++++++++++++----------- tests/integration/test_seed.py | 45 +++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 19 deletions(-) diff --git a/seed/__init__.py b/seed/__init__.py index e5795e4..b6e67c3 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -1,4 +1,6 @@ from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from itertools import groupby from sqlalchemy import ( Column, @@ -21,6 +23,14 @@ _PG_SEQUENCE = table("pg_sequence", Column("seqrelid"), Column("seqincrement")) +@dataclass(frozen=True) +class _SequencePosition: + column: Column[object] + sequence: str + increment: int + next_value: int + + def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: metadata = MetaData() metadata.reflect(bind=connection, only=list(fixtures)) @@ -31,8 +41,10 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: continue _reject_always_identity_values(seeded_table, rows) _lock_against_writers(connection, seeded_table) + before = _sequence_positions(connection, seeded_table) inserted[seeded_table.name] = _insert_rows(connection, seeded_table, rows) - _advance_sequences(connection, seeded_table) + for position in before: + _restart_past_stored_ids(connection, seeded_table, position) return inserted @@ -57,31 +69,30 @@ def _reject_always_identity_values(seeded_table: Table, rows: Sequence[Row]) -> def _insert_rows( connection: Connection, seeded_table: Table, rows: Sequence[Row] ) -> int: - batches: dict[frozenset[str], list[Row]] = {} - for row in rows: - batches.setdefault(frozenset(row), []).append(row) statement = insert(seeded_table).on_conflict_do_nothing().returning(*seeded_table.c) return sum( - len(connection.execute(statement, batch).all()) for batch in batches.values() + len(connection.execute(statement, list(batch)).all()) + for _, batch in groupby(rows, key=frozenset) ) -def _advance_sequences(connection: Connection, seeded_table: Table) -> None: +def _sequence_positions( + connection: Connection, seeded_table: Table +) -> list[_SequencePosition]: table_name = connection.dialect.identifier_preparer.format_table(seeded_table) + positions: list[_SequencePosition] = [] for column in seeded_table.c: sequence = connection.scalar( select(func.pg_get_serial_sequence(table_name, column.name)) ) if sequence is not None: - _advance_sequence(connection, seeded_table, column, sequence) + positions.append(_sequence_position(connection, column, sequence)) + return positions -def _advance_sequence( - connection: Connection, - seeded_table: Table, - column: Column[object], - sequence: str, -) -> None: +def _sequence_position( + connection: Connection, column: Column[object], sequence: str +) -> _SequencePosition: schema, name, increment = connection.execute( select( _PG_NAMESPACE.c.nspname, @@ -99,10 +110,21 @@ def _advance_sequence( select(state.c.last_value, state.c.is_called) ).one() next_value = last_value + increment if is_called else last_value - furthest = func.max if increment > 0 else func.min - stored = connection.scalar(select(furthest(column)).select_from(seeded_table)) - if stored is None or (next_value - stored) * increment > 0: - return - connection.execute( - text(f"ALTER SEQUENCE {sequence} RESTART WITH {int(stored + increment)}") + return _SequencePosition(column, sequence, increment, next_value) + + +def _restart_past_stored_ids( + connection: Connection, seeded_table: Table, before: _SequencePosition +) -> None: + furthest = func.max if before.increment > 0 else func.min + stored = connection.scalar( + select(furthest(before.column)).select_from(seeded_table) ) + target = before.next_value + if stored is not None and (stored - target) * before.increment >= 0: + target = stored + before.increment + after = _sequence_position(connection, before.column, before.sequence) + if target != after.next_value: + connection.execute( + text(f"ALTER SEQUENCE {before.sequence} RESTART WITH {int(target)}") + ) diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index ac111ba..fee6e50 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -98,6 +98,51 @@ def test_rows_with_different_columns_keep_their_own_values( 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_rerun_without_explicit_ids_does_not_consume_the_sequence( + connection: Connection, +) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " + "name text UNIQUE)" + ) + ) + fixtures = {"seed_probe": [{"name": "a"}, {"name": "b"}]} + + seed(connection, fixtures) + second = seed(connection, fixtures) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert second == {"seed_probe": 0} + assert new_id == 3 + + def test_descending_sequence_moves_past_the_lowest_id(connection: Connection) -> None: connection.execute( text( From 8339d6ff23272d57c456c64ff37bc34b84c2ed7d Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 23:20:06 +0530 Subject: [PATCH 07/13] Restart sequences on their increment progression --- seed/__init__.py | 3 ++- tests/integration/test_seed.py | 17 +++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/seed/__init__.py b/seed/__init__.py index b6e67c3..92ec100 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -122,7 +122,8 @@ def _restart_past_stored_ids( ) target = before.next_value if stored is not None and (stored - target) * before.increment >= 0: - target = stored + before.increment + steps = (stored - target) // before.increment + 1 + target += steps * before.increment after = _sequence_position(connection, before.column, before.sequence) if target != after.next_value: connection.execute( diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index fee6e50..e44af87 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -163,6 +163,23 @@ def test_descending_sequence_moves_past_the_lowest_id(connection: Connection) -> assert new_id == -6 +def test_sequence_keeps_its_increment_progression(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY " + "(START WITH 1 INCREMENT BY 2 MAXVALUE 3) PRIMARY KEY, name text)" + ) + ) + + seed(connection, {"seed_probe": [{"id": 2, "name": "seeded"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 3 + + def test_unused_sequence_keeps_its_configured_start(connection: Connection) -> None: connection.execute( text( From 4f1c65b5bfb86da8093237a9881cfa8e918d2820 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 23:29:54 +0530 Subject: [PATCH 08/13] Require a conflict key per fixture row and respect sequence bounds --- docs/local-development.md | 2 +- seed/__init__.py | 45 +++++++++++++++++++++++++++++++--- tests/integration/test_seed.py | 31 +++++++++++++++++++++++ 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/docs/local-development.md b/docs/local-development.md index 7be2059..c2cc646 100644 --- a/docs/local-development.md +++ b/docs/local-development.md @@ -49,7 +49,7 @@ 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`. Rows that already exist are kept as they are, so running the command again changes nothing. Explicit IDs need a serial or `GENERATED BY DEFAULT` identity column. After inserting them, the command moves the column's sequence past every stored ID, so the next insert gets a new one. +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 must set a primary key or unique key. Rows that already exist are kept as they are, so running the command again changes nothing. Explicit IDs need a serial or `GENERATED BY DEFAULT` identity column. After inserting them, the command moves the column's sequence past every stored ID, so the next insert gets a new one. ## Isolated stacks diff --git a/seed/__init__.py b/seed/__init__.py index 92ec100..37cd04f 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -7,6 +7,7 @@ Connection, MetaData, Table, + UniqueConstraint, cast, func, select, @@ -20,7 +21,13 @@ _PG_CLASS = table("pg_class", Column("oid"), Column("relname"), Column("relnamespace")) _PG_NAMESPACE = table("pg_namespace", Column("oid"), Column("nspname")) -_PG_SEQUENCE = table("pg_sequence", Column("seqrelid"), Column("seqincrement")) +_PG_SEQUENCE = table( + "pg_sequence", + Column("seqrelid"), + Column("seqincrement"), + Column("seqmin"), + Column("seqmax"), +) @dataclass(frozen=True) @@ -28,6 +35,8 @@ class _SequencePosition: column: Column[object] sequence: str increment: int + minimum: int + maximum: int next_value: int @@ -40,6 +49,7 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: if not rows: continue _reject_always_identity_values(seeded_table, rows) + _require_conflict_keys(seeded_table, rows) _lock_against_writers(connection, seeded_table) before = _sequence_positions(connection, seeded_table) inserted[seeded_table.name] = _insert_rows(connection, seeded_table, rows) @@ -66,6 +76,29 @@ def _reject_always_identity_values(seeded_table: Table, rows: Sequence[Row]) -> raise ValueError(msg) +def _require_conflict_keys(seeded_table: Table, rows: Sequence[Row]) -> None: + keys = [frozenset(column.name for column in seeded_table.primary_key.columns)] + keys += [ + frozenset(column.name for column in constraint.columns) + for constraint in seeded_table.constraints + if isinstance(constraint, UniqueConstraint) + ] + keys += [ + frozenset(column.name for column in index.columns) + for index in seeded_table.indexes + if index.unique and len(index.columns) == len(index.expressions) + ] + keys = [key for key in keys if key] + for row in rows: + present = {name for name, value in row.items() if value is not None} + if not any(key <= present for key in keys): + msg = ( + f"{seeded_table.name} fixture rows must set a primary key or unique " + "key, otherwise every run inserts them again" + ) + raise ValueError(msg) + + def _insert_rows( connection: Connection, seeded_table: Table, rows: Sequence[Row] ) -> int: @@ -93,11 +126,13 @@ def _sequence_positions( def _sequence_position( connection: Connection, column: Column[object], sequence: str ) -> _SequencePosition: - schema, name, increment = connection.execute( + schema, name, increment, minimum, maximum = connection.execute( select( _PG_NAMESPACE.c.nspname, _PG_CLASS.c.relname, _PG_SEQUENCE.c.seqincrement, + _PG_SEQUENCE.c.seqmin, + _PG_SEQUENCE.c.seqmax, ) .join_from( _PG_CLASS, _PG_NAMESPACE, _PG_CLASS.c.relnamespace == _PG_NAMESPACE.c.oid @@ -110,7 +145,7 @@ def _sequence_position( select(state.c.last_value, state.c.is_called) ).one() next_value = last_value + increment if is_called else last_value - return _SequencePosition(column, sequence, increment, next_value) + return _SequencePosition(column, sequence, increment, minimum, maximum, next_value) def _restart_past_stored_ids( @@ -118,7 +153,9 @@ def _restart_past_stored_ids( ) -> None: furthest = func.max if before.increment > 0 else func.min stored = connection.scalar( - select(furthest(before.column)).select_from(seeded_table) + select(furthest(before.column)) + .select_from(seeded_table) + .where(before.column.between(before.minimum, before.maximum)) ) target = before.next_value if stored is not None and (stored - target) * before.increment >= 0: diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index e44af87..61d15ba 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -180,6 +180,37 @@ def test_sequence_keeps_its_increment_progression(connection: Connection) -> Non assert new_id == 3 +def test_ids_outside_the_sequence_range_do_not_move_it( + connection: Connection, +) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY (MAXVALUE 10) PRIMARY KEY, " + "name text)" + ) + ) + + seed(connection, {"seed_probe": [{"id": 100, "name": "seeded"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 1 + + +def test_rows_without_a_conflict_key_are_rejected(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" + ) + ) + + with pytest.raises(ValueError, match="must set a primary key or unique key"): + seed(connection, {"seed_probe": [{"name": "a"}]}) + + def test_unused_sequence_keeps_its_configured_start(connection: Connection) -> None: connection.execute( text( From bf48606c9754afa3b1b0d5ac7ae0edda5bec5c56 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 23:39:14 +0530 Subject: [PATCH 09/13] Ignore partial indexes as seed keys and keep restarts in range --- seed/__init__.py | 8 ++++++-- tests/integration/test_seed.py | 35 ++++++++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/seed/__init__.py b/seed/__init__.py index 37cd04f..4410093 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -86,7 +86,9 @@ def _require_conflict_keys(seeded_table: Table, rows: Sequence[Row]) -> None: keys += [ frozenset(column.name for column in index.columns) for index in seeded_table.indexes - if index.unique and len(index.columns) == len(index.expressions) + if index.unique + and len(index.columns) == len(index.expressions) + and index.dialect_options["postgresql"]["where"] is None ] keys = [key for key in keys if key] for row in rows: @@ -160,7 +162,9 @@ def _restart_past_stored_ids( target = before.next_value if stored is not None and (stored - target) * before.increment >= 0: steps = (stored - target) // before.increment + 1 - target += steps * before.increment + advanced = target + steps * before.increment + if before.minimum <= advanced <= before.maximum: + target = advanced after = _sequence_position(connection, before.column, before.sequence) if target != after.next_value: connection.execute( diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index 61d15ba..4121301 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -211,6 +211,41 @@ def test_rows_without_a_conflict_key_are_rejected(connection: Connection) -> Non seed(connection, {"seed_probe": [{"name": "a"}]}) +def test_id_at_the_sequence_bound_keeps_the_seed_valid( + connection: Connection, +) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY (MAXVALUE 3) PRIMARY KEY, " + "name text)" + ) + ) + + seed(connection, {"seed_probe": [{"id": 3, "name": "seeded"}]}) + new_id = connection.scalar( + text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + ) + + assert new_id == 1 + + +def test_partial_unique_index_is_not_a_conflict_key(connection: Connection) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " + "email text, active boolean)" + ) + ) + connection.execute( + text("CREATE UNIQUE INDEX seed_probe_email ON seed_probe (email) WHERE active") + ) + + with pytest.raises(ValueError, match="must set a primary key or unique key"): + seed(connection, {"seed_probe": [{"email": "a", "active": False}]}) + + def test_unused_sequence_keeps_its_configured_start(connection: Connection) -> None: connection.execute( text( From 8459dead6011871b3ff1adf542bb1093381a5aa4 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Fri, 2 Oct 2026 23:49:01 +0530 Subject: [PATCH 10/13] Reserve explicit fixture IDs and reject sequence exhaustion --- seed/__init__.py | 45 ++++++++++++++++++++++++---------- tests/integration/test_seed.py | 32 ++++++++++++++++++------ 2 files changed, 57 insertions(+), 20 deletions(-) diff --git a/seed/__init__.py b/seed/__init__.py index 4410093..10586ed 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -52,9 +52,11 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: _require_conflict_keys(seeded_table, rows) _lock_against_writers(connection, seeded_table) before = _sequence_positions(connection, seeded_table) + for position in before: + _restart_past_ids(connection, seeded_table, position, rows) inserted[seeded_table.name] = _insert_rows(connection, seeded_table, rows) for position in before: - _restart_past_stored_ids(connection, seeded_table, position) + _restart_past_ids(connection, seeded_table, position, ()) return inserted @@ -150,23 +152,40 @@ def _sequence_position( return _SequencePosition(column, sequence, increment, minimum, maximum, next_value) -def _restart_past_stored_ids( - connection: Connection, seeded_table: Table, before: _SequencePosition +def _restart_past_ids( + connection: Connection, + seeded_table: Table, + before: _SequencePosition, + rows: Sequence[Row], ) -> None: - furthest = func.max if before.increment > 0 else func.min + column = before.column + ascending = before.increment > 0 stored = connection.scalar( - select(furthest(before.column)) + select(func.max(column) if ascending else func.min(column)) .select_from(seeded_table) - .where(before.column.between(before.minimum, before.maximum)) + .where(column.between(before.minimum, before.maximum)) ) + ids = [ + value + for row in rows + if isinstance(value := row.get(column.name), int) + and before.minimum <= value <= before.maximum + ] + if stored is not None: + ids.append(stored) target = before.next_value - if stored is not None and (stored - target) * before.increment >= 0: - steps = (stored - target) // before.increment + 1 - advanced = target + steps * before.increment - if before.minimum <= advanced <= before.maximum: - target = advanced - after = _sequence_position(connection, before.column, before.sequence) - if target != after.next_value: + if ids: + reached = max(ids) if ascending else min(ids) + if (reached - target) * before.increment >= 0: + target += ((reached - target) // before.increment + 1) * before.increment + if not before.minimum <= target <= before.maximum: + msg = ( + f"{seeded_table.name}.{column.name} has no sequence value left " + f"after ID {reached}" + ) + raise ValueError(msg) + current = _sequence_position(connection, column, before.sequence) + if target != current.next_value: connection.execute( text(f"ALTER SEQUENCE {before.sequence} RESTART WITH {int(target)}") ) diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index 4121301..f0eb17e 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -211,9 +211,7 @@ def test_rows_without_a_conflict_key_are_rejected(connection: Connection) -> Non seed(connection, {"seed_probe": [{"name": "a"}]}) -def test_id_at_the_sequence_bound_keeps_the_seed_valid( - connection: Connection, -) -> None: +def test_ids_that_exhaust_the_sequence_are_rejected(connection: Connection) -> None: connection.execute( text( "CREATE TABLE seed_probe (" @@ -222,12 +220,32 @@ def test_id_at_the_sequence_bound_keeps_the_seed_valid( ) ) - seed(connection, {"seed_probe": [{"id": 3, "name": "seeded"}]}) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") + with pytest.raises(ValueError, match="no sequence value left after ID 3"): + seed( + connection, + {"seed_probe": [{"id": 1, "name": "a"}, {"id": 3, "name": "b"}]}, + ) + + +def test_generated_rows_do_not_take_explicit_fixture_ids( + connection: Connection, +) -> None: + connection.execute( + text( + "CREATE TABLE seed_probe (" + "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " + "name text UNIQUE)" + ) ) - assert new_id == 1 + inserted = seed( + connection, + {"seed_probe": [{"name": "generated"}, {"id": 1, "name": "explicit"}]}, + ) + rows = connection.execute(text("SELECT id, name FROM seed_probe ORDER BY id")) + + assert inserted == {"seed_probe": 2} + assert [tuple(row) for row in rows] == [(1, "explicit"), (2, "generated")] def test_partial_unique_index_is_not_a_conflict_key(connection: Connection) -> None: From bb050d73651df3c5421595739440334b2fd27d23 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Sat, 3 Oct 2026 10:24:59 +0530 Subject: [PATCH 11/13] Narrow seeding to primary-key fixtures --- docs/local-development.md | 2 +- seed/__init__.py | 171 +++--------------------- tests/integration/test_seed.py | 236 +++------------------------------ 3 files changed, 40 insertions(+), 369 deletions(-) diff --git a/docs/local-development.md b/docs/local-development.md index c2cc646..6569b1a 100644 --- a/docs/local-development.md +++ b/docs/local-development.md @@ -49,7 +49,7 @@ 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 must set a primary key or unique key. Rows that already exist are kept as they are, so running the command again changes nothing. Explicit IDs need a serial or `GENERATED BY DEFAULT` identity column. After inserting them, the command moves the column's sequence past every stored ID, so the next insert gets a new one. +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 restarts the sequence after the highest stored ID, so the next insert gets a new one. This assumes the ascending, step-1 sequence that SQLAlchemy's `Identity()` creates. ## Isolated stacks diff --git a/seed/__init__.py b/seed/__init__.py index 10586ed..852e7ca 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -1,44 +1,12 @@ from collections.abc import Mapping, Sequence -from dataclasses import dataclass from itertools import groupby -from sqlalchemy import ( - Column, - Connection, - MetaData, - Table, - UniqueConstraint, - cast, - func, - select, - table, - text, -) -from sqlalchemy.dialects.postgresql import REGCLASS, insert +from sqlalchemy import Connection, MetaData, Table, func, select, text +from sqlalchemy.dialects.postgresql import 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")) -_PG_SEQUENCE = table( - "pg_sequence", - Column("seqrelid"), - Column("seqincrement"), - Column("seqmin"), - Column("seqmax"), -) - - -@dataclass(frozen=True) -class _SequencePosition: - column: Column[object] - sequence: str - increment: int - minimum: int - maximum: int - next_value: int - def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: metadata = MetaData() @@ -48,144 +16,47 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: rows = fixtures.get(seeded_table.name) if not rows: continue - _reject_always_identity_values(seeded_table, rows) - _require_conflict_keys(seeded_table, rows) + _require_primary_keys(seeded_table, rows) _lock_against_writers(connection, seeded_table) - before = _sequence_positions(connection, seeded_table) - for position in before: - _restart_past_ids(connection, seeded_table, position, rows) inserted[seeded_table.name] = _insert_rows(connection, seeded_table, rows) - for position in before: - _restart_past_ids(connection, seeded_table, position, ()) + _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 _reject_always_identity_values(seeded_table: Table, rows: Sequence[Row]) -> None: - for column in seeded_table.c: - identity = column.identity - if identity is None or not identity.always: - continue - if any(column.name in row for row in rows): - msg = ( - f"{seeded_table.name}.{column.name} is GENERATED ALWAYS, so fixtures " - "cannot set it; use GENERATED BY DEFAULT for seeded IDs" - ) - raise ValueError(msg) - - -def _require_conflict_keys(seeded_table: Table, rows: Sequence[Row]) -> None: - keys = [frozenset(column.name for column in seeded_table.primary_key.columns)] - keys += [ - frozenset(column.name for column in constraint.columns) - for constraint in seeded_table.constraints - if isinstance(constraint, UniqueConstraint) - ] - keys += [ - frozenset(column.name for column in index.columns) - for index in seeded_table.indexes - if index.unique - and len(index.columns) == len(index.expressions) - and index.dialect_options["postgresql"]["where"] is None - ] - keys = [key for key in keys if key] - for row in rows: - present = {name for name, value in row.items() if value is not None} - if not any(key <= present for key in keys): - msg = ( - f"{seeded_table.name} fixture rows must set a primary key or unique " - "key, otherwise every run inserts them again" - ) - raise ValueError(msg) - - def _insert_rows( connection: Connection, seeded_table: Table, rows: Sequence[Row] ) -> int: - statement = insert(seeded_table).on_conflict_do_nothing().returning(*seeded_table.c) + 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 _sequence_positions( - connection: Connection, seeded_table: Table -) -> list[_SequencePosition]: +def _restart_identity_sequences(connection: Connection, seeded_table: Table) -> None: table_name = connection.dialect.identifier_preparer.format_table(seeded_table) - positions: list[_SequencePosition] = [] - for column in seeded_table.c: + for column in seeded_table.primary_key.columns: sequence = connection.scalar( select(func.pg_get_serial_sequence(table_name, column.name)) ) - if sequence is not None: - positions.append(_sequence_position(connection, column, sequence)) - return positions - - -def _sequence_position( - connection: Connection, column: Column[object], sequence: str -) -> _SequencePosition: - schema, name, increment, minimum, maximum = connection.execute( - select( - _PG_NAMESPACE.c.nspname, - _PG_CLASS.c.relname, - _PG_SEQUENCE.c.seqincrement, - _PG_SEQUENCE.c.seqmin, - _PG_SEQUENCE.c.seqmax, - ) - .join_from( - _PG_CLASS, _PG_NAMESPACE, _PG_CLASS.c.relnamespace == _PG_NAMESPACE.c.oid - ) - .join(_PG_SEQUENCE, _PG_SEQUENCE.c.seqrelid == _PG_CLASS.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() - next_value = last_value + increment if is_called else last_value - return _SequencePosition(column, sequence, increment, minimum, maximum, next_value) - - -def _restart_past_ids( - connection: Connection, - seeded_table: Table, - before: _SequencePosition, - rows: Sequence[Row], -) -> None: - column = before.column - ascending = before.increment > 0 - stored = connection.scalar( - select(func.max(column) if ascending else func.min(column)) - .select_from(seeded_table) - .where(column.between(before.minimum, before.maximum)) - ) - ids = [ - value - for row in rows - if isinstance(value := row.get(column.name), int) - and before.minimum <= value <= before.maximum - ] - if stored is not None: - ids.append(stored) - target = before.next_value - if ids: - reached = max(ids) if ascending else min(ids) - if (reached - target) * before.increment >= 0: - target += ((reached - target) // before.increment + 1) * before.increment - if not before.minimum <= target <= before.maximum: - msg = ( - f"{seeded_table.name}.{column.name} has no sequence value left " - f"after ID {reached}" - ) - raise ValueError(msg) - current = _sequence_position(connection, column, before.sequence) - if target != current.next_value: + if sequence is None: + continue + highest = connection.execute(select(func.max(column))).scalar_one() connection.execute( - text(f"ALTER SEQUENCE {before.sequence} RESTART WITH {int(target)}") + text(f"ALTER SEQUENCE {sequence} RESTART WITH {int(highest) + 1}") ) diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index f0eb17e..a1a9724 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -13,6 +13,11 @@ 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]: @@ -43,13 +48,9 @@ def test_seeding_twice_changes_nothing(connection: Connection) -> None: def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" - ) - ) + 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") @@ -58,18 +59,11 @@ def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> assert new_id == 8 -def test_explicit_ids_for_always_generated_identity_are_rejected( - connection: Connection, -) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY, name text)" - ) - ) +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"seed_probe\.id is GENERATED ALWAYS"): - seed(connection, {"seed_probe": [{"id": 1, "name": "a"}]}) + 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( @@ -121,177 +115,11 @@ def test_rows_are_inserted_in_fixture_order(connection: Connection) -> None: assert inserted == {"seed_probe": 3} -def test_rerun_without_explicit_ids_does_not_consume_the_sequence( - connection: Connection, -) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " - "name text UNIQUE)" - ) - ) - fixtures = {"seed_probe": [{"name": "a"}, {"name": "b"}]} - - seed(connection, fixtures) - second = seed(connection, fixtures) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") - ) - - assert second == {"seed_probe": 0} - assert new_id == 3 - - -def test_descending_sequence_moves_past_the_lowest_id(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY " - "(INCREMENT BY -1 START WITH -1 MAXVALUE -1) PRIMARY KEY, name text)" - ) - ) - connection.execute( - text("INSERT INTO seed_probe (name) SELECT 'x' FROM generate_series(1, 2)") - ) - - seed(connection, {"seed_probe": [{"id": -5, "name": "seeded"}]}) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") - ) - - assert new_id == -6 - - -def test_sequence_keeps_its_increment_progression(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY " - "(START WITH 1 INCREMENT BY 2 MAXVALUE 3) PRIMARY KEY, name text)" - ) - ) - - seed(connection, {"seed_probe": [{"id": 2, "name": "seeded"}]}) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") - ) - - assert new_id == 3 - - -def test_ids_outside_the_sequence_range_do_not_move_it( - connection: Connection, -) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY (MAXVALUE 10) PRIMARY KEY, " - "name text)" - ) - ) - - seed(connection, {"seed_probe": [{"id": 100, "name": "seeded"}]}) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") - ) - - assert new_id == 1 - - -def test_rows_without_a_conflict_key_are_rejected(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" - ) - ) - - with pytest.raises(ValueError, match="must set a primary key or unique key"): - seed(connection, {"seed_probe": [{"name": "a"}]}) - - -def test_ids_that_exhaust_the_sequence_are_rejected(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY (MAXVALUE 3) PRIMARY KEY, " - "name text)" - ) - ) - - with pytest.raises(ValueError, match="no sequence value left after ID 3"): - seed( - connection, - {"seed_probe": [{"id": 1, "name": "a"}, {"id": 3, "name": "b"}]}, - ) - - -def test_generated_rows_do_not_take_explicit_fixture_ids( - connection: Connection, -) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " - "name text UNIQUE)" - ) - ) - - inserted = seed( - connection, - {"seed_probe": [{"name": "generated"}, {"id": 1, "name": "explicit"}]}, - ) - rows = connection.execute(text("SELECT id, name FROM seed_probe ORDER BY id")) - - assert inserted == {"seed_probe": 2} - assert [tuple(row) for row in rows] == [(1, "explicit"), (2, "generated")] - - -def test_partial_unique_index_is_not_a_conflict_key(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " - "email text, active boolean)" - ) - ) - connection.execute( - text("CREATE UNIQUE INDEX seed_probe_email ON seed_probe (email) WHERE active") - ) - - with pytest.raises(ValueError, match="must set a primary key or unique key"): - seed(connection, {"seed_probe": [{"email": "a", "active": False}]}) - - -def test_unused_sequence_keeps_its_configured_start(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY " - "(START WITH 100 MINVALUE 100) PRIMARY KEY, name text)" - ) - ) - - seed(connection, {"seed_probe": [{"id": 1, "name": "seeded"}]}) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") - ) - - assert new_id == 100 - - 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( - "CREATE TABLE seed_rollback_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " - "name text)" - ) - ) + 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() @@ -310,60 +138,32 @@ def test_rolled_back_seed_leaves_the_sequence_alone(test_postgres_url: str) -> N assert new_id == 1 -def test_sequence_never_moves_backwards(connection: Connection) -> None: - connection.execute( - text( - "CREATE TABLE seed_probe (" - "id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)" - ) - ) - connection.execute( - text("INSERT INTO seed_probe (name) SELECT 'x' FROM generate_series(1, 20)") - ) - connection.execute(text("DELETE FROM seed_probe WHERE id > 2")) - - seed(connection, {"seed_probe": [{"id": 3, "name": "seeded"}]}) - new_id = connection.scalar( - text("INSERT INTO seed_probe (name) VALUES ('c') RETURNING id") - ) - - assert new_id == 21 - - 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 GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " - "name text)" - ) + 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( - text("INSERT INTO seed_lock_probe (name) VALUES ('writer')") - ) + writer.execute(insert_row) writer.rollback() seeding.commit() - new_id = writer.scalar( - text( - "INSERT INTO seed_lock_probe (name) VALUES ('writer') RETURNING id" - ) - ) + 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 new_id == 6 + assert rows == 2 async def test_application_startup_does_not_seed( From bad4f9532fbb853b65d30e151556a54463e22b5f Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Sat, 3 Oct 2026 10:35:52 +0530 Subject: [PATCH 12/13] Seed only fixture tables in the default schema --- seed/__init__.py | 2 +- tests/integration/test_seed.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/seed/__init__.py b/seed/__init__.py index 852e7ca..5d1e02a 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -14,7 +14,7 @@ def seed(connection: Connection, fixtures: Fixtures) -> dict[str, int]: inserted: dict[str, int] = {} for seeded_table in metadata.sorted_tables: rows = fixtures.get(seeded_table.name) - if not rows: + if seeded_table.schema is not None or not rows: continue _require_primary_keys(seeded_table, rows) _lock_against_writers(connection, seeded_table) diff --git a/tests/integration/test_seed.py b/tests/integration/test_seed.py index a1a9724..abad042 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -115,6 +115,34 @@ def test_rows_are_inserted_in_fixture_order(connection: Connection) -> None: 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: From 662555e54b1f894abba532047c173a81255c652d Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Sat, 3 Oct 2026 10:43:55 +0530 Subject: [PATCH 13/13] Only advance identity sequences when seeding --- docs/local-development.md | 2 +- seed/__init__.py | 43 ++++++++++++++++++++++++++++------ tests/integration/test_seed.py | 16 +++++++++++++ 3 files changed, 53 insertions(+), 8 deletions(-) diff --git a/docs/local-development.md b/docs/local-development.md index 6569b1a..8eceff6 100644 --- a/docs/local-development.md +++ b/docs/local-development.md @@ -49,7 +49,7 @@ 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 restarts the sequence after the highest stored ID, so the next insert gets a new one. This assumes the ascending, step-1 sequence that SQLAlchemy's `Identity()` creates. +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 diff --git a/seed/__init__.py b/seed/__init__.py index 5d1e02a..5685ebb 100644 --- a/seed/__init__.py +++ b/seed/__init__.py @@ -1,12 +1,25 @@ from collections.abc import Mapping, Sequence from itertools import groupby -from sqlalchemy import Connection, MetaData, Table, func, select, text -from sqlalchemy.dialects.postgresql import insert +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() @@ -50,13 +63,29 @@ def _insert_rows( def _restart_identity_sequences(connection: Connection, seeded_table: Table) -> None: table_name = connection.dialect.identifier_preparer.format_table(seeded_table) - for column in seeded_table.primary_key.columns: + for key_column in seeded_table.primary_key.columns: sequence = connection.scalar( - select(func.pg_get_serial_sequence(table_name, column.name)) + select(func.pg_get_serial_sequence(table_name, key_column.name)) ) if sequence is None: continue - highest = connection.execute(select(func.max(column))).scalar_one() - connection.execute( - text(f"ALTER SEQUENCE {sequence} RESTART WITH {int(highest) + 1}") + 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/tests/integration/test_seed.py b/tests/integration/test_seed.py index abad042..bb148cc 100644 --- a/tests/integration/test_seed.py +++ b/tests/integration/test_seed.py @@ -59,6 +59,22 @@ def test_next_insert_after_explicit_ids_gets_a_new_id(connection: Connection) -> 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")))