luplo.core.qa ============= .. py:module:: luplo.core.qa .. autoapi-nested-parse:: QA check domain — a thin wrapper around items where ``item_type='qa_check'``. Tasks/items get verified by qa_checks. State changes initiated by users (start/pass/fail/block/skip/assign) are supersede operations; the revalidation trigger that demotes ``passed`` → ``pending`` after a target item gets superseded is in-place + audit (P8) and lives in ``LocalBackend.create_item`` — not here. Targets are stored as ``context.target_item_ids`` and ``context.target_task_ids`` arrays. GIN indexes from migration 0003 make reverse lookups fast. Attributes ---------- .. autoapisummary:: luplo.core.qa.ITEM_TYPE Functions --------- .. autoapisummary:: luplo.core.qa.create_qa luplo.core.qa.get_qa luplo.core.qa.list_qa luplo.core.qa.list_pending_for_task luplo.core.qa.list_pending_for_item luplo.core.qa.list_pending_for_wu luplo.core.qa.start_qa luplo.core.qa.pass_qa luplo.core.qa.fail_qa luplo.core.qa.block_qa luplo.core.qa.skip_qa luplo.core.qa.assign_qa Module Contents --------------- .. py:data:: ITEM_TYPE :value: 'qa_check' .. py:function:: create_qa(conn: psycopg.AsyncConnection[Any], *, project_id: str, title: str, actor_id: str, coverage: str, areas: list[str] | None = None, target_item_ids: list[str] | None = None, target_task_ids: list[str] | None = None, work_unit_id: str | None = None, body: str | None = None, context_extra: dict[str, Any] | None = None) -> luplo.core.models.Item :async: Create a new qa_check in ``pending`` status. *coverage* must be one of ``auto_partial`` / ``human_only``. .. py:function:: get_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, project_id: str | None = None) -> luplo.core.models.Item | None :async: Fetch the head of the chain containing *qa_id*. Accepts a full UUID or hex prefix (≥8 chars). Returns ``None`` if nothing matches; raises :class:`AmbiguousIdError` if a prefix resolves to multiple distinct chains. .. py:function:: list_qa(conn: psycopg.AsyncConnection[Any], project_id: str, *, status: str | None = None, work_unit_id: str | None = None) -> list[luplo.core.models.Item] :async: List qa_check heads in *project_id*, optionally filtered. .. py:function:: list_pending_for_task(conn: psycopg.AsyncConnection[Any], task_id: str) -> list[luplo.core.models.Item] :async: qa_checks targeting *task_id* that are not yet passed (GIN-hit). .. py:function:: list_pending_for_item(conn: psycopg.AsyncConnection[Any], item_id: str) -> list[luplo.core.models.Item] :async: qa_checks targeting *item_id* that are not yet passed (GIN-hit). .. py:function:: list_pending_for_wu(conn: psycopg.AsyncConnection[Any], work_unit_id: str) -> list[luplo.core.models.Item] :async: All pending qa_checks attached to *work_unit_id*. .. py:function:: start_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, actor_id: str, project_id: str | None = None) -> luplo.core.models.Item :async: .. py:function:: pass_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, actor_id: str, evidence: str | None = None, project_id: str | None = None) -> luplo.core.models.Item :async: .. py:function:: fail_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, actor_id: str, reason: str, project_id: str | None = None) -> luplo.core.models.Item :async: .. py:function:: block_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, actor_id: str, reason: str, project_id: str | None = None) -> luplo.core.models.Item :async: .. py:function:: skip_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, actor_id: str, project_id: str | None = None) -> luplo.core.models.Item :async: .. py:function:: assign_qa(conn: psycopg.AsyncConnection[Any], qa_id: str, *, actor_id: str, assignee_actor_id: str, project_id: str | None = None) -> luplo.core.models.Item :async: Assign a qa_check to *assignee_actor_id*. ``actor_id`` ≠ assignee by design (the actor performing the assignment is rarely the assignee).