state-hub/api/models/workplan_dependency.py

import uuid

from sqlalchemy import CheckConstraint, ForeignKey, Index, String, Text, text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship

from api.models.base import Base, TimestampMixin, new_uuid


class WorkplanDependency(Base, TimestampMixin):
    """Directed dependency edge: `from_workplan` depends on a workplan or task.

    Semantics: the target must reach a satisfactory state before `from_workplan`
    can fully proceed.  Hard deletes are intentional —
    removing an edge removes a constraint, not information.
    """

    __tablename__ = "workplan_dependencies"
    __table_args__ = (
        CheckConstraint(
            "(to_workplan_id IS NOT NULL AND to_task_id IS NULL) "
            "OR (to_workplan_id IS NULL AND to_task_id IS NOT NULL)",
            name="ck_wp_dep_exactly_one_target",
        ),
        Index(
            "uq_wp_dep_workplan_target",
            "from_workplan_id",
            "to_workplan_id",
            "relationship_type",
            unique=True,
            postgresql_where=text("to_workplan_id IS NOT NULL AND from_task_id IS NULL"),
        ),
        Index(
            "uq_wp_dep_task_target",
            "from_workplan_id",
            "to_task_id",
            "relationship_type",
            unique=True,
            postgresql_where=text("to_task_id IS NOT NULL AND from_task_id IS NULL"),
        ),
        # Task-block `depends_on` edges (CUST-WP-0074): the waiting task is the
        # from side, so two tasks in one workplan may wait on the same target.
        Index(
            "uq_wp_dep_from_task_workplan_target",
            "from_task_id",
            "to_workplan_id",
            "relationship_type",
            unique=True,
            postgresql_where=text("to_workplan_id IS NOT NULL AND from_task_id IS NOT NULL"),
        ),
        Index(
            "uq_wp_dep_from_task_task_target",
            "from_task_id",
            "to_task_id",
            "relationship_type",
            unique=True,
            postgresql_where=text("to_task_id IS NOT NULL AND from_task_id IS NOT NULL"),
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=new_uuid
    )
    from_workplan_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("workplans.id", ondelete="CASCADE", onupdate="CASCADE"),
        nullable=False,
        index=True,
    )
    # Set when the edge originates from a task block rather than workplan
    # frontmatter; always belongs to `from_workplan_id`.
    from_task_id: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("tasks.id", ondelete="CASCADE", onupdate="CASCADE"),
        nullable=True,
        index=True,
    )
    to_workplan_id: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("workplans.id", ondelete="CASCADE", onupdate="CASCADE"),
        nullable=True,
        index=True,
    )
    to_task_id: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("tasks.id", ondelete="CASCADE", onupdate="CASCADE"),
        nullable=True,
        index=True,
    )
    relationship_type: Mapped[str] = mapped_column(
        String(40), nullable=False, default="blocks", server_default="blocks", index=True
    )
    description: Mapped[str | None] = mapped_column(Text, nullable=True)

    from_workplan: Mapped["Workplan"] = relationship(  # noqa: F821
        "Workplan", foreign_keys=[from_workplan_id]
    )
    to_workplan: Mapped["Workplan | None"] = relationship(  # noqa: F821
        "Workplan", foreign_keys=[to_workplan_id]
    )
    from_task: Mapped["Task | None"] = relationship("Task", foreign_keys=[from_task_id])  # noqa: F821
    to_task: Mapped["Task | None"] = relationship("Task", foreign_keys=[to_task_id])  # noqa: F821