20260731_09_transactional_aigc_outbox.py 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. """turn the AIGC effect ledger into a transactional outbox
  2. Revision ID: 20260731_09
  3. Revises: 20260731_08
  4. Create Date: 2026-07-31
  5. """
  6. from __future__ import annotations
  7. from collections.abc import Sequence
  8. import sqlalchemy as sa
  9. from alembic import op
  10. revision: str = "20260731_09"
  11. down_revision: str | None = "20260731_08"
  12. branch_labels: str | Sequence[str] | None = None
  13. depends_on: str | Sequence[str] | None = None
  14. def upgrade() -> None:
  15. op.add_column(
  16. "pipeline_outbox",
  17. sa.Column(
  18. "attempt_count",
  19. sa.Integer(),
  20. nullable=False,
  21. server_default="0",
  22. ),
  23. )
  24. op.add_column(
  25. "pipeline_outbox",
  26. sa.Column("last_attempt_at", sa.DateTime(), nullable=True),
  27. )
  28. op.add_column(
  29. "pipeline_outbox",
  30. sa.Column("next_retry_at", sa.DateTime(), nullable=True),
  31. )
  32. op.add_column(
  33. "pipeline_outbox",
  34. sa.Column("external_id", sa.String(length=128), nullable=True),
  35. )
  36. op.add_column(
  37. "pipeline_outbox",
  38. sa.Column("external_name", sa.String(length=512), nullable=True),
  39. )
  40. op.add_column(
  41. "pipeline_outbox",
  42. sa.Column("response_json", sa.JSON(), nullable=True),
  43. )
  44. outbox = sa.table(
  45. "pipeline_outbox",
  46. sa.column("status", sa.String(length=32)),
  47. )
  48. # Rows produced by the old implementation were written only after the
  49. # external call completed, so they represent already-dispatched effects.
  50. op.execute(
  51. outbox.update()
  52. .where(outbox.c.status == "dispatched")
  53. .values(status="succeeded")
  54. )
  55. def downgrade() -> None:
  56. with op.batch_alter_table("pipeline_outbox") as batch_op:
  57. batch_op.drop_column("response_json")
  58. batch_op.drop_column("external_name")
  59. batch_op.drop_column("external_id")
  60. batch_op.drop_column("next_retry_at")
  61. batch_op.drop_column("last_attempt_at")
  62. batch_op.drop_column("attempt_count")