api/modules/checklists/models.py - master - systemsstrategyy/Treker

api/modules/checklists/models.py
288 строк13 KB
SystemsStrategy
Initial import
02 июл 2026, 15:15
02 июл 2026, 15:159624ed1
"""ORM-модели checklists-схемы (`app.checklist_*`).

Five tables по schema-request к Foundation:
- ChecklistTemplate / ChecklistTemplateItem — шаблоны и их пункты
- ChecklistRun / ChecklistRunItem — конкретное прохождение
- ChecklistSchedule — расписание авто-запуска

Все таблицы в Postgres schema `app` (создаётся baseline-миграцией
Foundation). Миграция этих таблиц генерируется Foundation через
`alembic revision --autogenerate` после мерджа schema-request issue.

Граничные решения:
- `space_id` (на app.spaces) и `card_id` (на app.cards) хранятся как
  `BigInteger` БЕЗ `ForeignKey` constraint — Kanban-таблиц ещё не
  существует. Проверка существования — в service-слое через
  `CardService.get_card_summary`. Follow-up миграция от Foundation
  добавит FK когда Kanban мерджит.
- `status` хранится как VARCHAR + CHECK constraint (как `auth.users.role`).
  Значения = `RunItemStatus` enum из `domain.py` — single source of truth.
- `next_run_at` денормализуется в schedules для partial index
  `WHERE is_active`, ускоряет Procrastinate periodic-task scan.
"""

from __future__ import annotations

from datetime import datetime

from sqlalchemy import (
    DDL,
    BigInteger,
    Boolean,
    CheckConstraint,
    DateTime,
    ForeignKey,
    Integer,
    String,
    event,
    func,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship

from api.modules.checklists.domain import RunItemStatus
from api.shared.db.base import Base, TimestampMixin, UUIDMixin

# Тестовые fixtures Foundation (`tests/modules/auth/conftest.py`)
# используют `Base.metadata.create_all` workaround вместо запуска
# alembic-миграций. Этот workaround создаёт только те schemas, которые
# fixture явно прописывает (`auth`). Когда мои модели попадают в `Base`,
# `create_all` пытается выполнить `CREATE TABLE app.checklist_*` без
# существующей schema `app` — DDL падает.
#
# Решение: подписываемся на событие `before_create` у общего MetaData
# и создаём schema `app` идемпотентно ровно перед DDL-проходом. В prod
# schema создаётся baseline-миграцией Foundation (b1a5e0000001), здесь
# мы покрываем только test-fixture-путь. конвенций разработки явно отмечает
# `Base.metadata.create_all` как slabое место — это и есть оно.
event.listen(
    Base.metadata,
    "before_create",
    DDL("CREATE SCHEMA IF NOT EXISTS app"),  # type: ignore[no-untyped-call]
)

# Список разрешённых значений status в БД-CHECK — собирается из enum,
# чтобы при добавлении нового статуса в RunItemStatus обновление шло
# в одном месте (single source of truth — конвенций разработки).
_STATUS_VALUES = tuple(s.value for s in RunItemStatus)
_STATUS_CHECK = f"status IN {_STATUS_VALUES!r}"
# XOR на assignee: ровно одно из user/group. Используется в runs и schedules.
_ASSIGNEE_XOR = "(assigned_to_user_id IS NULL) <> (assigned_to_group_id IS NULL)"

# ── Templates ────────────────────────────────────────────────────────────────


class ChecklistTemplate(UUIDMixin, TimestampMixin, Base):
    """Шаблон контрольного листа — переиспользуемый набор пунктов.

    Архивирование через `archived_at` (мягкое) сохраняет ссылки из
    истории прохождений — `DELETE` запрещён.
    """

    __tablename__ = "checklist_templates"
    __table_args__ = {"schema": "app"}

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    # space_id — soft reference на app.spaces.id (Kanban). Без FK
    # constraint до тех пор пока Kanban не смержит свою миграцию.
    space_id: Mapped[int] = mapped_column(BigInteger, nullable=False, index=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str] = mapped_column(
        String(2000), nullable=False, default="", server_default=""
    )
    archived_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    # Per-template настраиваемый deadline в часах. Если set — run'ы из
    # этого шаблона получают `expected_completion_at = started_at + N`.
    # NULL → fallback на global 24h в worker'е `check_overdue_runs`.
    # Migration в alembic — issue #139 (Foundation).
    default_duration_hours: Mapped[int | None] = mapped_column(Integer, nullable=True)

    items: Mapped[list[ChecklistTemplateItem]] = relationship(
        back_populates="template",
        cascade="all, delete-orphan",
        order_by="ChecklistTemplateItem.sort_order",
    )


class ChecklistTemplateItem(UUIDMixin, TimestampMixin, Base):
    """Пункт шаблона. `required=True` — пункт обязателен в прохождении
    (если будет skipped, run не считается successfully completed)."""

    __tablename__ = "checklist_template_items"
    __table_args__ = ({"schema": "app"},)

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    template_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("app.checklist_templates.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    text: Mapped[str] = mapped_column(String(500), nullable=False)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    required: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true"
    )
    archived_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    template: Mapped[ChecklistTemplate] = relationship(back_populates="items")


# ── Runs ─────────────────────────────────────────────────────────────────────


class ChecklistRun(UUIDMixin, TimestampMixin, Base):
    """Конкретное прохождение шаблона.

    Может быть привязан к карточке (`card_id IS NOT NULL`) или standalone
    (для регулярных аудитов отдела). Назначение — XOR на юзера или группу.

    Append-only после `completed_at IS NOT NULL`: items нельзя изменять.
    Это даёт audit-доказательства для compliance-отчётов (конвенций разработки
    «История как audit log»).
    """

    __tablename__ = "checklist_runs"
    __table_args__ = (
        CheckConstraint(_ASSIGNEE_XOR, name="ck_checklist_runs_assignee_xor"),
        {"schema": "app"},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    template_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("app.checklist_templates.id", ondelete="RESTRICT"),
        nullable=False,
        index=True,
    )
    # card_id — soft reference к app.cards.id (Kanban). Без FK constraint.
    card_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True, index=True)
    started_by: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("auth.users.id", ondelete="RESTRICT"),
        nullable=False,
    )
    started_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    assigned_to_user_id: Mapped[int | None] = mapped_column(
        BigInteger,
        ForeignKey("auth.users.id", ondelete="SET NULL"),
        nullable=True,
    )
    assigned_to_group_id: Mapped[int | None] = mapped_column(
        BigInteger,
        ForeignKey("auth.user_groups.id", ondelete="SET NULL"),
        nullable=True,
    )
    expected_completion_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    # Persistent dedup marker для periodic-task `check_overdue_runs`:
    # после отправки overdue-уведомления ставится в now(), повторные сканы
    # фильтруют по `IS NULL`. См. issue #114.
    overdue_notified_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )

    template: Mapped[ChecklistTemplate] = relationship()
    items: Mapped[list[ChecklistRunItem]] = relationship(
        back_populates="run",
        cascade="all, delete-orphan",
    )


class ChecklistRunItem(UUIDMixin, TimestampMixin, Base):
    """Состояние одного пункта в run'е. Статус управляется через
    `domain.transition_item` — БД enforces только enum-значения."""

    __tablename__ = "checklist_run_items"
    __table_args__ = (
        CheckConstraint(_STATUS_CHECK, name="ck_checklist_run_items_status"),
        {"schema": "app"},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    run_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("app.checklist_runs.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    # Имя FK задаём явно — иначе автоматическое
    # `fk_checklist_run_items_template_item_id_checklist_template_items` = 64 char,
    # Postgres лимит = 63. Это байт, который Quality Agent в своих unit-тестах
    # не поймал (там не делается реальный CREATE TABLE).
    template_item_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey(
            "app.checklist_template_items.id",
            ondelete="RESTRICT",
            name="fk_run_items_template_item_id",
        ),
        nullable=False,
    )
    status: Mapped[str] = mapped_column(
        String(20),
        nullable=False,
        default=RunItemStatus.PENDING.value,
        server_default=RunItemStatus.PENDING.value,
    )
    comment: Mapped[str | None] = mapped_column(String(2000), nullable=True)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    completed_by: Mapped[int | None] = mapped_column(
        BigInteger,
        ForeignKey("auth.users.id", ondelete="SET NULL"),
        nullable=True,
    )

    run: Mapped[ChecklistRun] = relationship(back_populates="items")
    template_item: Mapped[ChecklistTemplateItem] = relationship()


# ── Schedules ────────────────────────────────────────────────────────────────


class ChecklistSchedule(UUIDMixin, TimestampMixin, Base):
    """Расписание автоматического запуска шаблона.

    Procrastinate periodic-task сканирует
        SELECT * FROM checklist_schedules
        WHERE is_active AND next_run_at <= now()
    раз в минуту. `next_run_at` денормализуется чтобы partial index
    `WHERE is_active` ускорял scan.

    XOR на assignee такой же как у runs. Удаление юзера/группы каскадно
    удаляет их расписания (некому исполнять — нет смысла хранить).
    """

    __tablename__ = "checklist_schedules"
    __table_args__ = (
        CheckConstraint(_ASSIGNEE_XOR, name="ck_checklist_schedules_assignee_xor"),
        {"schema": "app"},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    template_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("app.checklist_templates.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    cron_expr: Mapped[str] = mapped_column(String(100), nullable=False)
    assigned_to_user_id: Mapped[int | None] = mapped_column(
        BigInteger,
        ForeignKey("auth.users.id", ondelete="CASCADE"),
        nullable=True,
    )
    assigned_to_group_id: Mapped[int | None] = mapped_column(
        BigInteger,
        ForeignKey("auth.user_groups.id", ondelete="CASCADE"),
        nullable=True,
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true"
    )
    next_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)