paper-doc/backend/app/crud/template.py

"""Data access for templates and their ordered field selections."""

from collections.abc import Sequence

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from app.crud.filters import LIKE_ESCAPE, like_pattern
from app.models import Template, TemplateFieldLibrary, TemplateField
from app.schemas.template import TemplateListItem, TemplateFieldInput


def _conditions(keyword: str | None) -> list:
    """Search both the name and the abstract from one box."""
    if not keyword:
        return []
    pattern = like_pattern(keyword)
    return [
        or_(
            Template.name.like(pattern, escape=LIKE_ESCAPE),
            Template.abstract.like(pattern, escape=LIKE_ESCAPE),
        )
    ]


def _field_count_column():
    """A correlated ``COUNT`` of the template's placement rows."""
    return (
        select(func.count(TemplateField.id))
        .where(TemplateField.template_id == Template.id)
        .correlate(Template)
        .scalar_subquery()
    )


def list_templates(
    db: Session,
    *,
    keyword: str | None = None,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[TemplateListItem], int]:
    """Return one page of templates with their field counts, plus the total.

    The outline itself is left out: a table row needs the count, not the rows.
    Recently edited templates come first, since that is what a user returns to.
    """
    conditions = _conditions(keyword)

    total = db.scalar(
        select(func.count(Template.id)).where(*conditions)
    ) or 0

    stmt = (
        select(Template, _field_count_column().label("field_count"))
        .where(*conditions)
        .order_by(Template.updated_at.desc(), Template.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )

    items = [
        TemplateListItem(
            id=template.id,
            name=template.name,
            abstract=template.abstract,
            field_count=field_count,
            created_at=template.created_at,
            updated_at=template.updated_at,
        )
        for template, field_count in db.execute(stmt).all()
    ]
    return items, total


def get(db: Session, template_id: int) -> Template | None:
    """Return one template with its outline already loaded, or ``None``.

    ``items`` (ordered by ``sort``) and each item's ``field`` are both
    configured for eager loading on the relationships, so this is a fixed
    number of queries rather than one per field.
    """
    return db.get(Template, template_id)


def name_taken(db: Session, name: str, *, exclude_id: int | None = None) -> bool:
    """Whether another template already uses ``name``."""
    stmt = select(func.count(Template.id)).where(Template.name == name)
    if exclude_id is not None:
        stmt = stmt.where(Template.id != exclude_id)
    return bool(db.scalar(stmt))


def missing_field_ids(db: Session, field_ids: Sequence[int]) -> list[int]:
    """Which of ``field_ids`` do not exist in the field library.

    Returning the offenders rather than a boolean lets the API name them, which
    is the difference between a usable error and "invalid request".
    """
    wanted = set(field_ids)
    if not wanted:
        return []
    found = set(
        db.scalars(select(TemplateFieldLibrary.id).where(TemplateFieldLibrary.id.in_(wanted))).all()
    )
    return sorted(wanted - found)


def _build_items(fields: Sequence[TemplateFieldInput]) -> list[TemplateField]:
    """Materialise the client's selection into placement rows."""
    return [TemplateField(field_id=item.field_id, sort=item.sort) for item in fields]


def create(
    db: Session,
    *,
    name: str,
    abstract: str | None,
    fields: Sequence[TemplateFieldInput],
) -> Template:
    """Insert a template together with its ordered selection."""
    template = Template(name=name, abstract=abstract)
    template.items = _build_items(fields)
    db.add(template)
    db.commit()
    db.refresh(template)
    return template


def update(
    db: Session,
    template: Template,
    *,
    name: str | None = None,
    abstract: str | None = None,
    abstract_provided: bool = False,
    fields: Sequence[TemplateFieldInput] | None = None,
) -> Template:
    """Apply a partial update. ``fields=None`` leaves the selection untouched.

    ``abstract_provided`` distinguishes "clear the abstract" from "leave it" —
    both arrive as ``None`` in the payload, and only the caller knows which the
    client meant.
    """
    if name is not None:
        template.name = name
    if abstract_provided:
        template.abstract = abstract
    if fields is not None:
        # delete-orphan removes the dropped rows on flush. TiDB does not
        # enforce ON DELETE CASCADE, so this ORM-level cascade is the only
        # thing cleaning up the join table.
        template.items.clear()
        db.flush()
        template.items.extend(_build_items(fields))

    db.commit()
    db.refresh(template)
    return template


def delete(db: Session, template: Template) -> None:
    """Delete a template and its placement rows."""
    db.delete(template)
    db.commit()


def delete_many(db: Session, template_ids: Sequence[int]) -> int:
    """Delete several templates, returning how many actually existed."""
    if not template_ids:
        return 0
    templates = list(
        db.scalars(
            select(Template).where(Template.id.in_(list(template_ids)))
        ).all()
    )
    for template in templates:
        db.delete(template)
    db.commit()
    return len(templates)