| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071 |
- """turn the AIGC effect ledger into a transactional outbox
- Revision ID: 20260731_09
- Revises: 20260731_08
- Create Date: 2026-07-31
- """
- from __future__ import annotations
- from collections.abc import Sequence
- import sqlalchemy as sa
- from alembic import op
- revision: str = "20260731_09"
- down_revision: str | None = "20260731_08"
- branch_labels: str | Sequence[str] | None = None
- depends_on: str | Sequence[str] | None = None
- def upgrade() -> None:
- op.add_column(
- "pipeline_outbox",
- sa.Column(
- "attempt_count",
- sa.Integer(),
- nullable=False,
- server_default="0",
- ),
- )
- op.add_column(
- "pipeline_outbox",
- sa.Column("last_attempt_at", sa.DateTime(), nullable=True),
- )
- op.add_column(
- "pipeline_outbox",
- sa.Column("next_retry_at", sa.DateTime(), nullable=True),
- )
- op.add_column(
- "pipeline_outbox",
- sa.Column("external_id", sa.String(length=128), nullable=True),
- )
- op.add_column(
- "pipeline_outbox",
- sa.Column("external_name", sa.String(length=512), nullable=True),
- )
- op.add_column(
- "pipeline_outbox",
- sa.Column("response_json", sa.JSON(), nullable=True),
- )
- outbox = sa.table(
- "pipeline_outbox",
- sa.column("status", sa.String(length=32)),
- )
- # Rows produced by the old implementation were written only after the
- # external call completed, so they represent already-dispatched effects.
- op.execute(
- outbox.update()
- .where(outbox.c.status == "dispatched")
- .values(status="succeeded")
- )
- def downgrade() -> None:
- with op.batch_alter_table("pipeline_outbox") as batch_op:
- batch_op.drop_column("response_json")
- batch_op.drop_column("external_name")
- batch_op.drop_column("external_id")
- batch_op.drop_column("next_retry_at")
- batch_op.drop_column("last_attempt_at")
- batch_op.drop_column("attempt_count")
|