SupplyAgent/supply_infra/db/models/pipeline_outbox.py

from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy import (
    Boolean,
    DateTime,
    ForeignKey,
    Index,
    JSON,
    String,
    Text,
    UniqueConstraint,
    func,
)
from sqlalchemy.orm import Mapped, mapped_column

from supply_infra.db.base import Base


class PipelineOutbox(Base):
    """Ledger of dispatched AIGC write effects."""

    __tablename__ = "pipeline_outbox"
    __table_args__ = (
        UniqueConstraint("idempotency_key", name="uk_pipeline_outbox_idempotency"),
        Index("idx_pipeline_outbox_run", "run_id", "step_run_id"),
        Index("idx_pipeline_outbox_status", "status", "created_at"),
    )

    outbox_id: Mapped[str] = mapped_column(String(36), primary_key=True)
    run_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("pipeline_run.run_id", ondelete="CASCADE"),
        nullable=False,
    )
    step_run_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("pipeline_step_run.step_run_id", ondelete="CASCADE"),
        nullable=False,
    )
    effect_type: Mapped[str] = mapped_column(String(64), nullable=False)
    idempotency_key: Mapped[str] = mapped_column(String(191), nullable=False)
    payload_hash: Mapped[str] = mapped_column(String(64), nullable=False)
    payload_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    payload_uri: Mapped[str | None] = mapped_column(String(1024), nullable=True)
    status: Mapped[str] = mapped_column(
        String(32),
        nullable=False,
        default="dispatched",
    )
    dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    record_error: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=func.now(),
        onupdate=func.now(),
    )