diff --git a/docs/agents.md b/docs/agents.md index 32aca9a..331c60e 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -366,7 +366,7 @@ handler it installs, for when you assemble the pipeline yourself. Abstract ORM bases — bind them to your own `DeclarativeBase`: `OutboxEventDBBase`, `InboxEventDBBase`, `OutboxEventPartitionedDBBase`, `InboxEventPartitionedDBBase`, the mixin `EventMixin`, and the helpers `get_event_constraints(table_name, -include_created_at_in_unique=False)` and `UnConstrainedEnum`. +include_created_at_in_unique=False, *, metadata=None)` and `UnConstrainedEnum`. `PostgresOutboxRepository(session, *, model_class, conflict_index_id=None, conflict_index_idempotency=None, batch_size=1000, error_max_length=2000, @@ -488,7 +488,11 @@ protocols, not a third implementation. `EventBatchProcessor` sets Give the sink an idempotent key. 17. **The tables are yours.** No `Base`, no migrations, no DDL. Bind the abstract bases to your `DeclarativeBase` and generate the migration yourself; the repositories depend on the - column names, so keep them. + column names, so keep them. The check constraints are named `ck__` unless + your `MetaData` has a `ck` naming convention with `%(constraint_name)s` in it — then the + bases hand it the bare rule (`attempts_valid`, …) and the convention names them, so nothing + runs past PostgreSQL's 63 bytes. Composing `__table_args__` yourself, pass + `metadata=Base.metadata` to `get_event_constraints` for the same. 18. **`shutdown_requested_func` stops a batch, it does not abort an event.** It is polled before the fetch — returning `True` there locks nothing at all — and again before each event. What was already processed is committed; the events left untouched come back in diff --git a/docs/api_reference.md b/docs/api_reference.md index 001b034..04bbe1d 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -197,7 +197,7 @@ All inherit from `OmniBoxError`. - ORM bases: `OutboxEventDBBase`, `InboxEventDBBase`, `OutboxEventPartitionedDBBase`, `InboxEventPartitionedDBBase`, plus the underlying `EventMixin`, `OutboxColumnsMixin`, `InboxColumnsMixin`. - Repositories: `PostgresOutboxRepository`, `PostgresInboxRepository`, `PostgresEventRepository` (shared base). Both expose `session` — the `AsyncSession` they were built on, read-only — which is how a handler passed to `InboxConsumerRunner` writes its side effects in the transaction that inserts the inbox row. -- Helpers: `UnConstrainedEnum`, `get_event_constraints(table_name, include_created_at_in_unique=False)`. +- Helpers: `UnConstrainedEnum`, `get_event_constraints(table_name, include_created_at_in_unique=False, *, metadata=None)`. ### Kafka (extra: `kafka`) diff --git a/docs/migrations.md b/docs/migrations.md index 7048961..b62e45d 100644 --- a/docs/migrations.md +++ b/docs/migrations.md @@ -205,6 +205,28 @@ class InboxEventDB(Base, InboxEventDBBase): You can override `__tablename__`, add service-specific columns, or change `__inbox_dedup_index_columns__` if you partition. +## Naming conventions + +The check constraints are named `ck_
_` — `ck_outbox_events_attempts_valid`, as in the DDL above — unless the `MetaData` of your `DeclarativeBase` carries a `ck` naming convention that interpolates `%(constraint_name)s`. Then the bases hand the convention the bare rule (`attempts_valid`, `completed_status_consistency`, `lock_consistency`) and it builds the name: `sqlalchemy-foundation-kit`'s `"%(table_name)s_%(constraint_name)s_check"` gives `outbox_events_attempts_valid_check`, SQLAlchemy's documented `"ck_%(table_name)s_%(constraint_name)s"` gives `ck_outbox_events_attempts_valid` again. Every name fits PostgreSQL's 63-byte limit either way; the longest, `outbox_events_partitioned_completed_status_consistency_check`, is 60. A finished name handed to such a convention would be qualified a second time, run past 63 bytes and be truncated with a hash. The indexes keep their `idx_
_…` names under any convention — SQLAlchemy leaves an explicitly named `Index` alone unless the `ix` rule interpolates `%(constraint_name)s`, which neither of the conventions above does. + +If you compose `__table_args__` yourself, pass the metadata so the helper can see the convention: + +```python +class OrdersOutbox(Base, OutboxEventDBBase): + __tablename__ = "orders_outbox" + __table_args__ = get_event_constraints("orders_outbox", metadata=Base.metadata) +``` + +Up to 0.2.1 the bases always declared the finished name, so a database created under such a convention holds the doubly qualified names, six of the twelve truncated with a hash — `outbox_events_ck_outbox_events_completed_status_consist_a784`. Alembic's autogenerate does not compare check constraints unless `alembic.ext.checkconstraint_byname` (Alembic 1.19+) is enabled, so rename them in a hand-written migration. They are the ones with `_ck_` in the middle: + +```sql +SELECT conrelid::regclass, conname FROM pg_constraint WHERE contype = 'c' AND conname LIKE '%\_ck\_%'; +ALTER TABLE outbox_events RENAME CONSTRAINT outbox_events_ck_outbox_events_completed_status_consist_a784 + TO outbox_events_completed_status_consistency_check; +``` + +The four bases also build `__table_args__` in a `declared_attr` now, so a workaround that read the tuple off the abstract class — `OutboxEventPartitionedDBBase.__table_args__`, to wrap the names in `conv()` by hand — raises `AttributeError: type object 'OutboxEventPartitionedDBBase' has no attribute 'metadata'` at import. Delete it; the bases produce the convention's names on their own. + ## Alembic example ```python diff --git a/omni_box/infra/storage/postgres/orm.py b/omni_box/infra/storage/postgres/orm.py index 0fe1b7b..504d33b 100644 --- a/omni_box/infra/storage/postgres/orm.py +++ b/omni_box/infra/storage/postgres/orm.py @@ -4,6 +4,7 @@ import datetime from functools import partial +from typing import ClassVar from uuid import UUID from sqlalchemy import ( @@ -11,11 +12,12 @@ DateTime, Enum, Index, + MetaData, String, text, ) from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.orm import Mapped, declared_attr, mapped_column from sqlalchemy.sql import func from ....core.constants import DEFAULT_MAX_ATTEMPTS @@ -36,6 +38,10 @@ class EventMixin: __abstract__ = True + # Supplied by the DeclarativeBase a concrete model is bound to; declared so that + # __table_args__ can read its naming convention. + metadata: ClassVar[MetaData] + # Identifiers id: Mapped[UUID] = mapped_column(primary_key=True) event_type: Mapped[str] = mapped_column(String(100), nullable=False) @@ -100,25 +106,45 @@ class InboxColumnsMixin: source: Mapped[str] = mapped_column(String(255), nullable=False) -def get_event_constraints(table_name: str, include_created_at_in_unique: bool = False) -> tuple: - """Generate generic constraints and indexes for event table.""" +def get_event_constraints( + table_name: str, include_created_at_in_unique: bool = False, *, metadata: MetaData | None = None +) -> tuple: + """Generate generic constraints and indexes for event table. + + Args: + table_name: Prefix of the constraint and index names. + include_created_at_in_unique: Add ``created_at`` to the unique index on ``idempotency_key`` + (a partitioned table requires the partition key in every unique index). + metadata: The ``MetaData`` the table is registered in. When its ``ck`` naming convention + contains ``%(constraint_name)s`` the check constraints are named by their rule alone + (``attempts_valid``, ``completed_status_consistency``, ``lock_consistency``) and the + convention qualifies them; otherwise they are named ``ck__``. + """ idempotency_key_cols = ["idempotency_key"] if include_created_at_in_unique: idempotency_key_cols.append("created_at") + # SQLAlchemy applies a convention to an explicitly named constraint only when the rule + # interpolates that name (sqlalchemy/sql/naming.py); a finished name would be qualified twice. + ck_rule = metadata.naming_convention.get("ck") if metadata is not None else None + convention_qualifies = isinstance(ck_rule, str) and "constraint_name" in ck_rule + + def check_name(rule: str) -> str: + return rule if convention_qualifies else f"ck_{table_name}_{rule}" + return ( # Ensure attempts_made never exceeds max_attempts at database level - CheckConstraint("attempts_made <= max_attempts", name=f"ck_{table_name}_attempts_valid"), + CheckConstraint("attempts_made <= max_attempts", name=check_name("attempts_valid")), # Status consistency constraints CheckConstraint( f"(status = '{EventStatus.COMPLETED.value}' AND completed_at IS NOT NULL) OR " f"(status != '{EventStatus.COMPLETED.value}' AND completed_at IS NULL)", - name=f"ck_{table_name}_completed_status_consistency", + name=check_name("completed_status_consistency"), ), # Locking consistency: locked_at and locked_by must be both NULL or both NOT NULL CheckConstraint( "(locked_at IS NULL AND locked_by IS NULL) OR (locked_at IS NOT NULL AND locked_by IS NOT NULL)", - name=f"ck_{table_name}_lock_consistency", + name=check_name("lock_consistency"), ), Index( f"idx_{table_name}_pending_fetch", @@ -153,7 +179,10 @@ class OutboxEventDBBase(EventMixin, OutboxColumnsMixin): __abstract__ = True __tablename__ = "outbox_events" - __table_args__ = get_event_constraints("outbox_events") + + @declared_attr.directive + def __table_args__(self) -> tuple: + return get_event_constraints("outbox_events", metadata=self.metadata) class InboxEventDBBase(EventMixin, InboxColumnsMixin): @@ -165,16 +194,18 @@ class InboxEventDBBase(EventMixin, InboxColumnsMixin): # Columns for INSERT ... ON CONFLICT DO NOTHING (PostgresInboxRepository) __inbox_dedup_index_columns__: tuple[str, ...] = ("message_id", "consumer_group") - __table_args__ = ( - *get_event_constraints("inbox_events"), - # Inbox specific unique constraint for deduplication - Index( - "idx_inbox_deduplication", - "message_id", - "consumer_group", - unique=True, - ), - ) + @declared_attr.directive + def __table_args__(self) -> tuple: + return ( + *get_event_constraints("inbox_events", metadata=self.metadata), + # Inbox specific unique constraint for deduplication + Index( + "idx_inbox_deduplication", + "message_id", + "consumer_group", + unique=True, + ), + ) class OutboxEventPartitionedDBBase(EventMixin, OutboxColumnsMixin): @@ -193,10 +224,12 @@ class OutboxEventPartitionedDBBase(EventMixin, OutboxColumnsMixin): __outbox_conflict_index_id__ = ("id", "created_at") __outbox_conflict_index_idempotency__ = ("idempotency_key", "created_at") - __table_args__ = ( - *get_event_constraints("outbox_events_p", include_created_at_in_unique=True), - {"postgresql_partition_by": "RANGE (created_at)"}, - ) + @declared_attr.directive + def __table_args__(self) -> tuple: + return ( + *get_event_constraints("outbox_events_p", include_created_at_in_unique=True, metadata=self.metadata), + {"postgresql_partition_by": "RANGE (created_at)"}, + ) class InboxEventPartitionedDBBase(EventMixin, InboxColumnsMixin): @@ -218,23 +251,25 @@ class InboxEventPartitionedDBBase(EventMixin, InboxColumnsMixin): __inbox_dedup_index_columns__: tuple[str, ...] = ("message_id", "consumer_group", "created_at") - __table_args__ = ( - *get_event_constraints("inbox_events_p", include_created_at_in_unique=True), - Index( - "idx_inbox_events_p_deduplication", - "message_id", - "consumer_group", - "created_at", - unique=True, - ), - Index( - "idx_inbox_events_p_message_consumer", - "message_id", - "consumer_group", - unique=False, - ), - {"postgresql_partition_by": "RANGE (created_at)"}, - ) + @declared_attr.directive + def __table_args__(self) -> tuple: + return ( + *get_event_constraints("inbox_events_p", include_created_at_in_unique=True, metadata=self.metadata), + Index( + "idx_inbox_events_p_deduplication", + "message_id", + "consumer_group", + "created_at", + unique=True, + ), + Index( + "idx_inbox_events_p_message_consumer", + "message_id", + "consumer_group", + unique=False, + ), + {"postgresql_partition_by": "RANGE (created_at)"}, + ) # Type aliases diff --git a/tests/integration/postgres/test_orm.py b/tests/integration/postgres/test_orm.py index 7014970..27b4f52 100644 --- a/tests/integration/postgres/test_orm.py +++ b/tests/integration/postgres/test_orm.py @@ -1,11 +1,30 @@ import pytest -from sqlalchemy import String, Table +from sqlalchemy import Connection, MetaData, String, Table, inspect, text +from sqlalchemy.ext.asyncio import AsyncEngine +from sqlalchemy.orm import DeclarativeBase from omni_box.core.models.enums import EventStatus +from omni_box.infra.storage.postgres.orm import ( + InboxEventDBBase, + InboxEventPartitionedDBBase, + OutboxEventDBBase, + OutboxEventPartitionedDBBase, +) from tests.models import ConcreteOutboxEvent pytestmark = pytest.mark.integration +CHECK_RULES = ("attempts_valid", "completed_status_consistency", "lock_consistency") + +# sqlalchemy-foundation-kit's DB_NAMING_CONVENTION +FOUNDATION_KIT_CONVENTION = { + "ix": "%(column_0_label)s_idx", + "uq": "%(table_name)s_%(column_0_name)s_key", + "ck": "%(table_name)s_%(constraint_name)s_check", + "fk": "%(table_name)s_%(column_0_name)s_fkey", + "pk": "%(table_name)s_pkey", +} + def test__outbox_event_db_model__concrete_model__has_expected_columns_and_defaults() -> None: # Arrange @@ -92,3 +111,45 @@ def test__outbox_event_db_model__concrete_model__has_expected_indexes() -> None: assert "scheduled_at" in [c.name for c in pending_fetch_idx.columns] assert pending_fetch_idx.dialect_options["postgresql"]["where"] is not None assert EventStatus.PENDING.value in str(pending_fetch_idx.dialect_options["postgresql"]["where"]) + + +async def test__event_bases__foundation_kit_convention__postgres_stores_the_qualified_names( + db_engine: AsyncEngine, +) -> None: + # Arrange + schema = "naming_convention" + + class Base(DeclarativeBase): + metadata = MetaData(naming_convention=FOUNDATION_KIT_CONVENTION, schema=schema) + + class Outbox(Base, OutboxEventDBBase): + pass + + class Inbox(Base, InboxEventDBBase): + pass + + class OutboxPartitioned(Base, OutboxEventPartitionedDBBase): + pass + + class InboxPartitioned(Base, InboxEventPartitionedDBBase): + pass + + def reflect_check_names(conn: Connection) -> dict[str, set[str]]: + inspector = inspect(conn) + return { + table.name: {c["name"] for c in inspector.get_check_constraints(table.name, schema=schema)} + for table in Base.metadata.sorted_tables + } + + # Act + async with db_engine.begin() as conn: + await conn.execute(text(f"CREATE SCHEMA {schema}")) + try: + await conn.run_sync(Base.metadata.create_all) + stored = await conn.run_sync(reflect_check_names) + finally: + await conn.execute(text(f"DROP SCHEMA {schema} CASCADE")) + + # Assert + for table in Base.metadata.sorted_tables: + assert stored[table.name] == {f"{table.name}_{rule}_check" for rule in CHECK_RULES} diff --git a/tests/unit/infra/storage/postgres/test_orm.py b/tests/unit/infra/storage/postgres/test_orm.py new file mode 100644 index 0000000..451b191 --- /dev/null +++ b/tests/unit/infra/storage/postgres/test_orm.py @@ -0,0 +1,178 @@ +"""Unit tests for the names ``get_event_constraints`` gives the event tables. + +A check constraint is declared with the finished name ``ck_
_`` unless the +``MetaData`` carries a ``ck`` naming convention that interpolates ``%(constraint_name)s``; +then the bases hand the convention the bare rule and it builds the name. A finished name +under such a convention is qualified twice, runs past PostgreSQL's 63-byte limit and is +truncated with a hash (issue #36). These tests compile the DDL through the PostgreSQL +dialect and never touch a database; the catalog round trip lives in +``tests/integration/postgres/test_orm.py``. +""" + +from __future__ import annotations + +import re + +import pytest +from sqlalchemy import CheckConstraint, MetaData, Table +from sqlalchemy.dialects import postgresql +from sqlalchemy.orm import DeclarativeBase +from sqlalchemy.schema import CreateTable + +from omni_box.infra.storage.postgres.orm import ( + InboxEventDBBase, + InboxEventPartitionedDBBase, + OutboxEventDBBase, + OutboxEventPartitionedDBBase, + get_event_constraints, +) + +pytestmark = pytest.mark.unit + +PG_MAX_IDENTIFIER = 63 +CHECK_RULES = ("attempts_valid", "completed_status_consistency", "lock_consistency") + +# sqlalchemy-foundation-kit's DB_NAMING_CONVENTION +FOUNDATION_KIT_CONVENTION = { + "ix": "%(column_0_label)s_idx", + "uq": "%(table_name)s_%(column_0_name)s_key", + "ck": "%(table_name)s_%(constraint_name)s_check", + "fk": "%(table_name)s_%(column_0_name)s_fkey", + "pk": "%(table_name)s_pkey", +} +# The convention SQLAlchemy's documentation recommends +SQLALCHEMY_DOCS_CONVENTION = { + "ix": "ix_%(column_0_label)s", + "uq": "uq_%(table_name)s_%(column_0_name)s", + "ck": "ck_%(table_name)s_%(constraint_name)s", + "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", + "pk": "pk_%(table_name)s", +} + +# (abstract base, the prefix its __table_args__ passes to get_event_constraints) +EVENT_BASES = ( + (OutboxEventDBBase, "outbox_events"), + (InboxEventDBBase, "inbox_events"), + (OutboxEventPartitionedDBBase, "outbox_events_p"), + (InboxEventPartitionedDBBase, "inbox_events_p"), +) + + +def _bind(naming_convention: dict[str, str] | None) -> dict[str, Table]: + """Bind the four abstract bases to a fresh DeclarativeBase and return its tables by name.""" + + class Base(DeclarativeBase): + metadata = MetaData(naming_convention=naming_convention) + + for base, _ in EVENT_BASES: + type(f"Concrete{base.__name__}", (Base, base), {}) + return {table.name: table for table in Base.metadata.sorted_tables} + + +def _check_names(table: Table) -> set[str]: + return {c.name for c in table.constraints if isinstance(c, CheckConstraint)} + + +def _ddl_check_names(table: Table) -> set[str]: + ddl = str(CreateTable(table).compile(dialect=postgresql.dialect())) + return set(re.findall(r"CONSTRAINT (\S+) CHECK", ddl)) + + +def _index_names(table: Table) -> set[str]: + return {index.name for index in table.indexes} + + +def test__event_bases__no_naming_convention__check_names_are_the_documented_ones() -> None: + # Act + tables = _bind(None) + + # Assert + for base, prefix in EVENT_BASES: + table = tables[base.__tablename__] + assert _check_names(table) == {f"ck_{prefix}_{rule}" for rule in CHECK_RULES} + assert _ddl_check_names(table) == _check_names(table) + + +@pytest.mark.parametrize( + ("naming_convention", "expected"), + [ + pytest.param(FOUNDATION_KIT_CONVENTION, "{table}_{rule}_check", id="foundation-kit"), + pytest.param(SQLALCHEMY_DOCS_CONVENTION, "ck_{table}_{rule}", id="sqlalchemy-docs"), + ], +) +def test__event_bases__ck_convention__check_names_are_qualified_once_and_fit_postgres( + naming_convention: dict[str, str], expected: str +) -> None: + # Act + tables = _bind(naming_convention) + + # Assert + for base, _ in EVENT_BASES: + table = tables[base.__tablename__] + names = _check_names(table) + assert names == {expected.format(table=table.name, rule=rule) for rule in CHECK_RULES} + assert all(len(name) <= PG_MAX_IDENTIFIER for name in names) + assert _ddl_check_names(table) == names + + +@pytest.mark.parametrize( + "naming_convention", + [ + pytest.param(FOUNDATION_KIT_CONVENTION, id="foundation-kit"), + pytest.param(SQLALCHEMY_DOCS_CONVENTION, id="sqlalchemy-docs"), + ], +) +def test__event_bases__ix_convention__explicit_index_names_are_kept(naming_convention: dict[str, str]) -> None: + # Arrange + without_convention = _bind(None) + + # Act + tables = _bind(naming_convention) + + # Assert + for name, table in tables.items(): + assert _index_names(table) == _index_names(without_convention[name]) + + +def test__event_bases__two_declarative_bases__each_gets_its_own_table_args() -> None: + # Act + first = _bind(None) + second = _bind(None) + + # Assert + for base, _ in EVENT_BASES: + name = base.__tablename__ + assert first[name] is not second[name] + assert _check_names(second[name]) == _check_names(first[name]) + assert _index_names(second[name]) == _index_names(first[name]) + + +@pytest.mark.parametrize( + "metadata", + [ + pytest.param(None, id="none"), + pytest.param(MetaData(), id="default"), + pytest.param(MetaData(naming_convention={"ck": "%(table_name)s_%(column_0_name)s_check"}), id="ck-by-column"), + ], +) +def test__get_event_constraints__no_ck_rule_interpolating_the_name__check_names_are_finished( + metadata: MetaData | None, +) -> None: + # Act + constraints = get_event_constraints("custom_events", metadata=metadata) + + # Assert + names = {c.name for c in constraints if isinstance(c, CheckConstraint)} + assert names == {f"ck_custom_events_{rule}" for rule in CHECK_RULES} + + +def test__get_event_constraints__ck_rule_interpolating_the_name__check_names_are_the_bare_rules() -> None: + # Arrange + metadata = MetaData(naming_convention=FOUNDATION_KIT_CONVENTION) + + # Act + constraints = get_event_constraints("custom_events", metadata=metadata) + + # Assert + names = {c.name for c in constraints if isinstance(c, CheckConstraint)} + assert names == set(CHECK_RULES)