b7ca7285cc
- services/file_status.py: set_file_status_for_job(session, job, status, *, now, only_from=None) и пакетный set_file_status_for_jobs для вахтёра; переход только при current_recognition_job_id = job.id, PROCESSED и SUPERSEDED не перезаписываются (D-5, R-5). - inbox: ack -> файл UPLOADED->PROCESSING; error -> ERROR; done/needs_check -> UPLOADED->PROCESSING (R-4: ack мог потеряться), из PROCESSING файл ждёт цикла маппинга; inbox не читает S3 и не создаёт документы. - вахтёр: UPDATE recognition_job ... RETURNING id и UPDATE stored_file -> ERROR в той же транзакции; публикатор при dlq -> файл ERROR. - удалены ResultMapper, NoopMapper, _map_result, _MapperError и параметр mapper у handle_message; app.py без mapper. - тесты: статусы файла (ack/error/done/needs_check/без ack/timeout/dlq), чужое задание файл не трогает (Review Focus 1), откат статуса файла вместе с транзакцией сообщения; CommittedJobs.make_file_job/add_job/file. - tasks.md 2.2, шаги Task 4 в плане. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
86 lines
4.0 KiB
Python
86 lines
4.0 KiB
Python
"""Статус `stored_file` по заданию распознавания (spec primary-files «Жизненный цикл статуса файла»; design D-5;
|
||
рулинги R-4, R-5; docs/context/05-domain-rules.md §9.2).
|
||
|
||
Единственное правило перехода: статус файла меняет только его ТЕКУЩЕЕ задание
|
||
(`stored_file.current_recognition_job_id = job.id`). Результат, ack, таймаут или dlq старого задания (повторная загрузка
|
||
после `SUPERSEDED`, будущий replay) файл не трогает. `PROCESSED` и `SUPERSEDED` — итог маппинга и автозамены — обменом
|
||
не перезаписываются.
|
||
|
||
Функции выполняют `UPDATE` в транзакции вызывающего (inbox, вахтёр, публикатор) — после изменения строки задания:
|
||
порядок блокировок «задание → файл» одинаков у всех участников, взаимоблокировок нет. `audit_event` не пишется:
|
||
обмен не аудируется (ASSUMPTION(OQ-55)).
|
||
"""
|
||
|
||
import uuid
|
||
from collections.abc import Collection, Sequence
|
||
from datetime import datetime
|
||
from typing import Any, Protocol
|
||
|
||
from sqlalchemy import CursorResult, update
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.models.enums import FileStatus
|
||
from src.models.files import StoredFile
|
||
|
||
# Итоги маппинга и автозамены: события обмена их не перезаписывают (design D-5)
|
||
PROTECTED_FILE_STATUSES: frozenset[FileStatus] = frozenset({FileStatus.PROCESSED, FileStatus.SUPERSEDED})
|
||
|
||
|
||
class JobRef(Protocol):
|
||
"""Что нужно от задания: `recognition_job.id` и `file_id` (ORM `RecognitionJob` или строка выборки)."""
|
||
|
||
@property
|
||
def id(self) -> uuid.UUID: ...
|
||
|
||
@property
|
||
def file_id(self) -> uuid.UUID: ...
|
||
|
||
|
||
def _rowcount(result: Any) -> int:
|
||
return result.rowcount if isinstance(result, CursorResult) else 0
|
||
|
||
|
||
async def set_file_status_for_job(
|
||
session: AsyncSession,
|
||
job: JobRef,
|
||
status: FileStatus,
|
||
*,
|
||
now: datetime,
|
||
only_from: Collection[FileStatus] | None = None,
|
||
) -> bool:
|
||
"""Перевести файл задания `job` в `status` (`updated_at = now`), если задание — текущее для файла и файл не в
|
||
`PROCESSED`/`SUPERSEDED`; `only_from` — допустимые исходные статусы (R-5: ack и результат без ack — только из
|
||
`UPLOADED`, запоздавший ack не воскрешает `ERROR`). Возвращает `True`, если строка файла изменена."""
|
||
stmt = (
|
||
update(StoredFile)
|
||
.where(
|
||
StoredFile.id == job.file_id,
|
||
StoredFile.current_recognition_job_id == job.id,
|
||
StoredFile.status.not_in(PROTECTED_FILE_STATUSES),
|
||
)
|
||
.values(status=status, updated_at=now)
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
if only_from is not None:
|
||
stmt = stmt.where(StoredFile.status.in_(list(only_from)))
|
||
return _rowcount(await session.execute(stmt)) == 1
|
||
|
||
|
||
async def set_file_status_for_jobs(
|
||
session: AsyncSession, job_ids: Sequence[uuid.UUID], status: FileStatus, *, now: datetime
|
||
) -> int:
|
||
"""Пакетный вариант для вахтёра: файлы, у которых текущее задание из `job_ids` и статус не `PROCESSED`/
|
||
`SUPERSEDED`, → `status`. Возвращает число изменённых файлов."""
|
||
if not job_ids:
|
||
return 0
|
||
stmt = (
|
||
update(StoredFile)
|
||
.where(
|
||
StoredFile.current_recognition_job_id.in_(list(job_ids)),
|
||
StoredFile.status.not_in(PROTECTED_FILE_STATUSES),
|
||
)
|
||
.values(status=status, updated_at=now)
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
return _rowcount(await session.execute(stmt))
|