api: загрузка файлов в консолидацию, переход версий «в работе»; query версии у загрузки ЮЛ удалён (#96)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
956a730a3c
commit
1e888d4670
@ -7,13 +7,14 @@ legal-entities/{legal_entity_id}/files` — рабочий стол консол
|
||||
import uuid
|
||||
from typing import Annotated, Any
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, status
|
||||
from fastapi import APIRouter, Depends, File, Query, Response, UploadFile, status
|
||||
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.files import PrimaryFilesServiceDep
|
||||
from src.api.endpoints.files.read import FileListQuery
|
||||
from src.api.endpoints.files.upload import accept_uploads
|
||||
from src.schemas.common import ErrorDetail, Page, PageParams
|
||||
from src.schemas.files import FileOut
|
||||
from src.schemas.files import FileOut, UploadResult
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@ -44,3 +45,40 @@ async def list_desk_files(
|
||||
consolidation_id, legal_entity_id, include_superseded=query.include_superseded, page=page
|
||||
)
|
||||
return Page[FileOut](items=items, total=total, limit=page.limit, offset=page.offset)
|
||||
|
||||
|
||||
UPLOAD_RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
**NOT_FOUND,
|
||||
status.HTTP_409_CONFLICT: {"model": ErrorDetail, "description": "CONSOLIDATION_NOT_EDITABLE — нет версии в работе"},
|
||||
status.HTTP_422_UNPROCESSABLE_CONTENT: {
|
||||
"model": UploadResult,
|
||||
"description": "ни один файл не принят; ЮЛ вне периметра версий в работе (ENTITY_NOT_IN_PERIMETER)",
|
||||
},
|
||||
status.HTTP_503_SERVICE_UNAVAILABLE: {"model": ErrorDetail, "description": "S3 не подключено"},
|
||||
}
|
||||
|
||||
|
||||
@router.post(
|
||||
DESK_PATH,
|
||||
operation_id="upload_consolidation_files", # имя тела multipart в OpenAPI без `__` (test_openapi_routes)
|
||||
response_model=UploadResult,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
dependencies=EDIT,
|
||||
responses=UPLOAD_RESPONSES,
|
||||
summary="Загрузка файлов ЮЛ в консолидацию",
|
||||
)
|
||||
async def upload_consolidation_files(
|
||||
consolidation_id: uuid.UUID,
|
||||
legal_entity_id: uuid.UUID,
|
||||
files: Annotated[list[UploadFile], File(description="Файлы ОСВ/ББ/ОФР: xls, xlsx, xlsb, pdf, png, jpg")],
|
||||
service: PrimaryFilesServiceDep,
|
||||
response: Response,
|
||||
) -> UploadResult:
|
||||
"""До приёма: консолидация и ЮЛ видимы, есть версия «в работе», ЮЛ в её периметре. Принятые файлы получают
|
||||
`consolidation_id`; после приёма хотя бы одного версии `INITIATED` с этим ЮЛ → `DOCUMENTS_PROCESSING`."""
|
||||
entity = await service.get_entity(legal_entity_id)
|
||||
consolidation = await service.check_consolidation_for_upload(consolidation_id, entity)
|
||||
result = await accept_uploads(service, entity, files, response, consolidation_id=consolidation.id)
|
||||
if result.accepted:
|
||||
await service.after_upload_to_consolidation(consolidation.id, entity.id)
|
||||
return result
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
"""`POST /legal-entities/{legal_entity_id}/files` — загрузка файлов организации (spec primary-files «Загрузка файлов
|
||||
организации», «Дедуп по содержимому»; design D-4).
|
||||
организации», «Дедуп по содержимому»; design D-4). Загрузка в консолидацию — `POST /consolidations/{id}/legal-entities/
|
||||
{id}/files` (`consolidation_files.py`), общий приём файлов — `accept_uploads`.
|
||||
|
||||
multipart: одно или несколько полей `files`; право `DOCUMENT_EDIT`; ЮЛ — в области видимости пользователя, иначе 404.
|
||||
Тип документа и период не запрашиваются (макет `40`). Каждый файл принимается или отклоняется отдельно (частичный
|
||||
@ -10,59 +11,30 @@ multipart: одно или несколько полей `files`; право `DO
|
||||
import uuid
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile, status
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status
|
||||
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.files import PrimaryFilesServiceDep
|
||||
from src.common.exceptions import DomainError
|
||||
from src.common.filenames import read_limited, safe_filename
|
||||
from src.models.directories import LegalEntity
|
||||
from src.schemas.common import ErrorDetail
|
||||
from src.schemas.files import AcceptedOut, RejectedOut, UploadResult
|
||||
from src.services.primary_files import FILE_TOO_LARGE, RejectedFile
|
||||
from src.services.primary_files import FILE_TOO_LARGE, PrimaryFilesService, RejectedFile
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.post(
|
||||
"/legal-entities/{legal_entity_id}/files",
|
||||
operation_id="upload_legal_entity_files", # имя тела multipart в OpenAPI без `__` (test_openapi_routes)
|
||||
response_model=UploadResult,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
dependencies=[Depends(require("DOCUMENT_EDIT"))],
|
||||
responses={
|
||||
status.HTTP_404_NOT_FOUND: {
|
||||
"model": ErrorDetail,
|
||||
"description": "ЮЛ нет или оно вне видимости; версия консолидации не найдена или чужая (VERSION_NOT_FOUND)",
|
||||
},
|
||||
status.HTTP_409_CONFLICT: {
|
||||
"model": ErrorDetail,
|
||||
"description": "версия консолидации не в работе (VERSION_NOT_EDITABLE)",
|
||||
},
|
||||
status.HTTP_422_UNPROCESSABLE_CONTENT: {
|
||||
"model": UploadResult,
|
||||
"description": "ни один файл не принят; ЮЛ вне периметра версии (ENTITY_NOT_IN_PERIMETER)",
|
||||
},
|
||||
status.HTTP_503_SERVICE_UNAVAILABLE: {"model": ErrorDetail, "description": "S3 не подключено"},
|
||||
},
|
||||
summary="Загрузка файлов организации",
|
||||
)
|
||||
async def upload_files(
|
||||
legal_entity_id: uuid.UUID,
|
||||
files: Annotated[list[UploadFile], File(description="Файлы ОСВ/ББ/ОФР: xls, xlsx, xlsb, pdf, png, jpg")],
|
||||
service: PrimaryFilesServiceDep,
|
||||
async def accept_uploads(
|
||||
service: PrimaryFilesService,
|
||||
entity: LegalEntity,
|
||||
files: list[UploadFile],
|
||||
response: Response,
|
||||
consolidation_version_id: Annotated[
|
||||
uuid.UUID | None,
|
||||
Query(description="Версия консолидации, в контексте которой загружаются файлы (необязательно)"),
|
||||
] = None,
|
||||
*,
|
||||
consolidation_id: uuid.UUID | None,
|
||||
) -> UploadResult:
|
||||
"""Принятые файлы (`file_id`, `job_id`, имя) и отклонённые (имя, код, пояснение, `file_id` дубля).
|
||||
|
||||
С `consolidation_version_id`: до приёма — версия видима, «в работе», ЮЛ в периметре; после приёма хотя бы одного
|
||||
файла версия `INITIATED` → `DOCUMENTS_PROCESSING` (отдельная транзакция, блокировка версии не нужна)."""
|
||||
entity = await service.get_entity(legal_entity_id)
|
||||
if consolidation_version_id is not None:
|
||||
await service.check_version_for_upload(consolidation_version_id, entity)
|
||||
"""Общий приём файлов двух маршрутов загрузки: каждый файл принимается или отклоняется отдельно; 422 — если не
|
||||
принят ни один. `consolidation_id` пишется в `stored_file` (загрузка в консолидацию) или `None`."""
|
||||
service.require_storage()
|
||||
accepted: list[AcceptedOut] = []
|
||||
rejected: list[RejectedOut] = []
|
||||
@ -78,7 +50,7 @@ async def upload_files(
|
||||
)
|
||||
continue
|
||||
try:
|
||||
result = await service.upload_one(entity, name, content)
|
||||
result = await service.upload_one(entity, name, content, consolidation_id=consolidation_id)
|
||||
except DomainError as error:
|
||||
failure = RejectedFile.from_error(safe_filename(name, default=""), error)
|
||||
rejected.append(
|
||||
@ -88,6 +60,28 @@ async def upload_files(
|
||||
accepted.append(AcceptedOut(file_id=result.file_id, job_id=result.job_id, name=result.name))
|
||||
if not accepted:
|
||||
response.status_code = status.HTTP_422_UNPROCESSABLE_CONTENT
|
||||
elif consolidation_version_id is not None:
|
||||
await service.after_upload_to_version(consolidation_version_id)
|
||||
return UploadResult(accepted=accepted, rejected=rejected)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/legal-entities/{legal_entity_id}/files",
|
||||
operation_id="upload_legal_entity_files", # имя тела multipart в OpenAPI без `__` (test_openapi_routes)
|
||||
response_model=UploadResult,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
dependencies=[Depends(require("DOCUMENT_EDIT"))],
|
||||
responses={
|
||||
status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "ЮЛ нет или оно вне видимости"},
|
||||
status.HTTP_422_UNPROCESSABLE_CONTENT: {"model": UploadResult, "description": "ни один файл не принят"},
|
||||
status.HTTP_503_SERVICE_UNAVAILABLE: {"model": ErrorDetail, "description": "S3 не подключено"},
|
||||
},
|
||||
summary="Загрузка файлов организации",
|
||||
)
|
||||
async def upload_files(
|
||||
legal_entity_id: uuid.UUID,
|
||||
files: Annotated[list[UploadFile], File(description="Файлы ОСВ/ББ/ОФР: xls, xlsx, xlsb, pdf, png, jpg")],
|
||||
service: PrimaryFilesServiceDep,
|
||||
response: Response,
|
||||
) -> UploadResult:
|
||||
"""Принятые файлы (`file_id`, `job_id`, имя) и отклонённые (имя, код, пояснение, `file_id` дубля)."""
|
||||
entity = await service.get_entity(legal_entity_id)
|
||||
return await accept_uploads(service, entity, files, response, consolidation_id=None)
|
||||
|
||||
@ -179,6 +179,42 @@ async def lock_versions_for_entity(
|
||||
return list((await session.execute(members_only)).scalars())
|
||||
|
||||
|
||||
async def versions_of(session: AsyncSession, consolidation_id: uuid.UUID) -> list[ConsolidationVersion]:
|
||||
"""Все версии консолидации по возрастанию `version_no` (без блокировки; проверки до приёма файлов, D-3)."""
|
||||
stmt = (
|
||||
select(ConsolidationVersion)
|
||||
.where(ConsolidationVersion.consolidation_id == consolidation_id)
|
||||
.order_by(ConsolidationVersion.version_no)
|
||||
)
|
||||
return list((await session.execute(stmt)).scalars())
|
||||
|
||||
|
||||
async def lock_versions_of_consolidation(
|
||||
session: AsyncSession,
|
||||
consolidation_id: uuid.UUID,
|
||||
entity_id: uuid.UUID,
|
||||
statuses: Collection[ConsolidationStatus],
|
||||
) -> list[ConsolidationVersion]:
|
||||
"""Версии консолидации со статусом из `statuses`, где ЮЛ — член периметра, под `FOR UPDATE ORDER BY id` (R-2;
|
||||
design D-4: переход после загрузки в консолидацию). `populate_existing` — свежий статус из БД."""
|
||||
is_member = exists().where(
|
||||
ConsolidationMember.consolidation_version_id == ConsolidationVersion.id,
|
||||
ConsolidationMember.legal_entity_id == entity_id,
|
||||
)
|
||||
stmt = (
|
||||
select(ConsolidationVersion)
|
||||
.where(
|
||||
ConsolidationVersion.consolidation_id == consolidation_id,
|
||||
ConsolidationVersion.status.in_(statuses),
|
||||
is_member,
|
||||
)
|
||||
.order_by(ConsolidationVersion.id)
|
||||
.with_for_update()
|
||||
.execution_options(populate_existing=True)
|
||||
)
|
||||
return list((await session.execute(stmt)).scalars())
|
||||
|
||||
|
||||
async def active_lock(session: AsyncSession, version_id: uuid.UUID) -> ConsolidationLock | None:
|
||||
"""Активная блокировка версии (`released_at IS NULL`; не более одной — partial UNIQUE)."""
|
||||
stmt = select(ConsolidationLock).where(
|
||||
|
||||
@ -8,8 +8,8 @@ audit-trail «События файлов и документов»; design D-4)
|
||||
1. рекомендательная блокировка `(ЮЛ, sha256)` до конца транзакции запроса и проверка дубля — до записи в S3;
|
||||
2. `s3_object` + `flush` (конфликт ключа или ошибка БД — до загрузки объекта, бакет не засоряется);
|
||||
3. `put_object`; ошибка S3 — `STORAGE_UNAVAILABLE`, SAVEPOINT откатывается (строки `s3_object` не остаётся);
|
||||
4. `stored_file` (`UPLOADED`, `version_no = 1`) и `recognition_job` (`queued`) + `flush`,
|
||||
`stored_file.current_recognition_job_id`, аудит `FILE_UPLOAD`.
|
||||
4. `stored_file` (`UPLOADED`, `version_no = 1`, `consolidation_id` при загрузке в консолидацию) и `recognition_job`
|
||||
(`queued`) + `flush`, `stored_file.current_recognition_job_id`, аудит `FILE_UPLOAD` (с `consolidation_id`).
|
||||
|
||||
Если после шага 3 упадёт БД (вставка файла/задания, аудит или commit запроса), объект в S3 останется без записи
|
||||
`s3_object` — допустимо: ключ содержит свежие `file_id`/`job_id`, ни на что не ссылается и не будет переиспользован;
|
||||
@ -37,7 +37,7 @@ from src.core.config import Settings
|
||||
from src.core.storage import ObjectNotFoundError, ObjectStorage
|
||||
from src.domain.file_types import MAX_ATTEMPTS, DetectedFile, detect_file
|
||||
from src.domain.policy import IN_WORK_STATUSES, VisibilityScope
|
||||
from src.models.consolidation import ConsolidationVersion
|
||||
from src.models.consolidation import Consolidation
|
||||
from src.models.directories import LegalEntity
|
||||
from src.models.enums import (
|
||||
AuditAction,
|
||||
@ -56,7 +56,7 @@ from src.schemas.common import PageParams
|
||||
from src.schemas.files import FileOut
|
||||
from src.services import consolidation_documents
|
||||
from src.services.audit import AuditService
|
||||
from src.services.consolidation_guard import CONSOLIDATION_NOT_FOUND, VERSION_NOT_FOUND
|
||||
from src.services.consolidation_guard import CONSOLIDATION_NOT_FOUND
|
||||
from src.services.documents import file_out
|
||||
from src.services.legal_entities import LEGAL_ENTITY_NOT_FOUND
|
||||
from src.services.result_mapping import RESULTS_PREFIX
|
||||
@ -160,34 +160,35 @@ class PrimaryFilesService:
|
||||
raise NotFoundError(LEGAL_ENTITY_NOT_FOUND)
|
||||
return entity
|
||||
|
||||
async def check_version_for_upload(self, version_id: uuid.UUID, entity: LegalEntity) -> ConsolidationVersion:
|
||||
"""Загрузка в контексте версии консолидации (spec primary-files «Загрузка файлов организации», D-6): ДО приёма
|
||||
файлов версия видима (иначе 404 `VERSION_NOT_FOUND`) → статус «в работе» (иначе 409 `VERSION_NOT_EDITABLE`) →
|
||||
ЮЛ — член периметра (иначе 422 `ENTITY_NOT_IN_PERIMETER`). Блокировка версии (`consolidation_lock`) и
|
||||
`FOR UPDATE` не нужны: статус перепроверяется под блокировкой в `after_upload_to_version`."""
|
||||
found = await consolidations_repo.get_version(self._session, self._scope, version_id)
|
||||
if found is None:
|
||||
raise NotFoundError(VERSION_NOT_FOUND)
|
||||
version = found[0]
|
||||
if version.status not in IN_WORK_STATUSES:
|
||||
raise ConflictError("VERSION_NOT_EDITABLE", f"Версия в статусе {version.status.value} не редактируется")
|
||||
members = await consolidations_repo.members(self._session, version.id)
|
||||
if all(member.legal_entity_id != entity.id for member, _ in members):
|
||||
raise DomainValidationError("ENTITY_NOT_IN_PERIMETER", "ЮЛ не входит в периметр версии консолидации")
|
||||
return version
|
||||
async def check_consolidation_for_upload(self, consolidation_id: uuid.UUID, entity: LegalEntity) -> Consolidation:
|
||||
"""Загрузка в консолидацию (spec primary-files «Загрузка файлов в консолидацию», §9.5 п. 8, D-3): ДО приёма —
|
||||
консолидация видима (404 `CONSOLIDATION_NOT_FOUND`) → есть версия «в работе» (409 `CONSOLIDATION_NOT_EDITABLE`)
|
||||
→ ЮЛ в периметре хотя бы одной из них (422 `ENTITY_NOT_IN_PERIMETER`). Блокировки не нужны: статус
|
||||
перечитывается под `FOR UPDATE` в `after_upload_to_consolidation`."""
|
||||
consolidation = await consolidations_repo.get_consolidation(self._session, self._scope, consolidation_id)
|
||||
if consolidation is None:
|
||||
raise NotFoundError(CONSOLIDATION_NOT_FOUND)
|
||||
in_work = [
|
||||
version
|
||||
for version in await consolidations_repo.versions_of(self._session, consolidation.id)
|
||||
if version.status in IN_WORK_STATUSES
|
||||
]
|
||||
if not in_work:
|
||||
raise ConflictError("CONSOLIDATION_NOT_EDITABLE", "У консолидации нет версии в работе")
|
||||
for version in in_work:
|
||||
members = await consolidations_repo.members(self._session, version.id)
|
||||
if any(member.legal_entity_id == entity.id for member, _ in members):
|
||||
return consolidation
|
||||
raise DomainValidationError("ENTITY_NOT_IN_PERIMETER", "ЮЛ не входит в периметр версий консолидации в работе")
|
||||
|
||||
async def after_upload_to_version(self, version_id: uuid.UUID) -> None:
|
||||
"""После приёма хотя бы одного файла: версия `INITIATED` → `DOCUMENTS_PROCESSING` (`changed_by` — загрузивший).
|
||||
async def after_upload_to_consolidation(self, consolidation_id: uuid.UUID, entity_id: uuid.UUID) -> None:
|
||||
"""После приёма хотя бы одного файла: каждая версия консолидации «в работе» с ЮЛ в периметре и статусом
|
||||
`INITIATED` → `DOCUMENTS_PROCESSING` (`changed_by` — загрузивший; §9.1, D-4).
|
||||
|
||||
В ОТДЕЛЬНОЙ транзакции (`session_transaction`, как аудит входа): `lock_version` (`FOR UPDATE`) → статус
|
||||
перечитывается — если уже не `INITIATED` (параллельный переход, аннулирование), ничего не меняется. Версия
|
||||
блокируется на короткое время и вне транзакции запроса, чтобы загрузка не держала её на время приёма файлов
|
||||
(D-9). Транзакция запроса к этому моменту версию не трогала, взаимных ожиданий нет.
|
||||
|
||||
Приём файлов важнее побочного статуса (R-10, design №43): любая ошибка перехода (`Exception`) логируется и
|
||||
НЕ меняет ответ — принятые файлы остаются, версия остаётся `INITIATED` до следующей загрузки или
|
||||
подтверждения. `BaseException` (`asyncio.CancelledError`) не перехватывается."""
|
||||
# импорт внутри функции — как в consolidation_documents (design №36): модули консолидации импортируют друг друга
|
||||
В ОТДЕЛЬНОЙ транзакции (`session_transaction`): версии под `FOR UPDATE ORDER BY id`, статус перечитан —
|
||||
попадают только ещё `INITIATED`. Приём файлов важнее побочного статуса (R-10 Среза 4): любая ошибка
|
||||
перехода логируется и НЕ меняет ответ. `BaseException` не перехватывается."""
|
||||
# импорт внутри функции — модули консолидации импортируют друг друга
|
||||
from src.services.consolidation_status import transition # noqa: PLC0415
|
||||
|
||||
factory = self._transaction
|
||||
@ -197,25 +198,28 @@ class PrimaryFilesService:
|
||||
factory = session_transaction
|
||||
try:
|
||||
async with factory() as session:
|
||||
version = await consolidations_repo.lock_version(session, version_id)
|
||||
if version is None or version.status is not ConsolidationStatus.INITIATED:
|
||||
return
|
||||
await transition(
|
||||
session,
|
||||
AuditService(session),
|
||||
version=version,
|
||||
to=ConsolidationStatus.DOCUMENTS_PROCESSING,
|
||||
changed_by=self._audit.user_id,
|
||||
comment=None,
|
||||
ip_address=self._audit.ip_address,
|
||||
user_agent=self._audit.user_agent,
|
||||
now=utc_now(),
|
||||
versions = await consolidations_repo.lock_versions_of_consolidation(
|
||||
session, consolidation_id, entity_id, {ConsolidationStatus.INITIATED}
|
||||
)
|
||||
audit = AuditService(session)
|
||||
now = utc_now()
|
||||
for version in versions:
|
||||
await transition(
|
||||
session,
|
||||
audit,
|
||||
version=version,
|
||||
to=ConsolidationStatus.DOCUMENTS_PROCESSING,
|
||||
changed_by=self._audit.user_id,
|
||||
comment=None,
|
||||
ip_address=self._audit.ip_address,
|
||||
user_agent=self._audit.user_agent,
|
||||
now=now,
|
||||
)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Не удалось перевести версию в DOCUMENTS_PROCESSING после загрузки, файлы приняты. "
|
||||
"error_category=version_transition_error, version_id={version_id}",
|
||||
version_id=version_id,
|
||||
"Не удалось перевести версии консолидации в DOCUMENTS_PROCESSING после загрузки, файлы приняты. "
|
||||
"error_category=version_transition_error, consolidation_id={consolidation_id}",
|
||||
consolidation_id=consolidation_id,
|
||||
)
|
||||
|
||||
def require_storage(self) -> ObjectStorage:
|
||||
@ -386,7 +390,9 @@ class PrimaryFilesService:
|
||||
)
|
||||
return (await self._session.execute(stmt)).scalar_one_or_none()
|
||||
|
||||
async def upload_one(self, entity: LegalEntity, name: str, content: bytes) -> AcceptedFile:
|
||||
async def upload_one(
|
||||
self, entity: LegalEntity, name: str, content: bytes, *, consolidation_id: uuid.UUID | None = None
|
||||
) -> AcceptedFile:
|
||||
"""Принять один файл (см. docstring модуля); отказ — `DomainError` (`UNSUPPORTED_FILE`, `DUPLICATE_FILE`,
|
||||
`STORAGE_UNAVAILABLE`), записи этого файла откатываются вместе с его SAVEPOINT."""
|
||||
storage = self.require_storage()
|
||||
@ -403,7 +409,13 @@ class PrimaryFilesService:
|
||||
file_id=str(duplicate.id),
|
||||
)
|
||||
return await self._store(
|
||||
entity, name=safe_name, content=content, sha256=sha256, storage=storage, detected=detected
|
||||
entity,
|
||||
name=safe_name,
|
||||
content=content,
|
||||
sha256=sha256,
|
||||
storage=storage,
|
||||
detected=detected,
|
||||
consolidation_id=consolidation_id,
|
||||
)
|
||||
|
||||
async def _lock_content(self, entity_id: uuid.UUID, sha256: str) -> None:
|
||||
@ -421,6 +433,7 @@ class PrimaryFilesService:
|
||||
sha256: str,
|
||||
storage: ObjectStorage,
|
||||
detected: DetectedFile,
|
||||
consolidation_id: uuid.UUID | None,
|
||||
) -> AcceptedFile:
|
||||
now = utc_now()
|
||||
user_id = self._audit.user_id
|
||||
@ -452,6 +465,7 @@ class PrimaryFilesService:
|
||||
stored_file = StoredFile(
|
||||
id=file_id,
|
||||
legal_entity_id=entity.id,
|
||||
consolidation_id=consolidation_id,
|
||||
file_type=detected.file_type,
|
||||
original_name=name,
|
||||
s3_object_id=s3_object.id,
|
||||
@ -491,6 +505,7 @@ class PrimaryFilesService:
|
||||
"content_hash": sha256,
|
||||
"job_id": str(job_id),
|
||||
"file_type": detected.file_type.value,
|
||||
"consolidation_id": str(consolidation_id) if consolidation_id is not None else None,
|
||||
},
|
||||
)
|
||||
return AcceptedFile(file_id=file_id, job_id=job_id, name=name)
|
||||
|
||||
@ -25,7 +25,7 @@ from src.api.dependencies.directories import get_app_settings
|
||||
from src.core.config import get_settings
|
||||
from src.models.access import AppUser
|
||||
from src.models.audit_ex import AuditEvent
|
||||
from src.models.consolidation import ConsolidationStatusHistory, ConsolidationVersion
|
||||
from src.models.consolidation import Consolidation, ConsolidationStatusHistory, ConsolidationVersion
|
||||
from src.models.directories import LegalEntity, RegionalOffice
|
||||
from src.models.enums import (
|
||||
AuditAction,
|
||||
@ -41,7 +41,7 @@ from src.models.enums import (
|
||||
from src.models.files import RecognitionJob, S3Object, StoredFile
|
||||
from src.services.recognition_exchange import publish_pending
|
||||
from tests.conftest import AuthClientFactory, FakeStorage
|
||||
from tests.factories import make_company_group, make_consolidation, make_legal_entity, make_lock
|
||||
from tests.factories import make_company_group, make_consolidation, make_legal_entity, make_lock, make_version
|
||||
from tests.test_recognition_exchange import REQUEST_VALIDATOR, FakeNats
|
||||
|
||||
HTTP_CREATED = 201
|
||||
@ -311,6 +311,7 @@ async def test_two_files_write_two_upload_events(
|
||||
"content_hash": hashlib.sha256(content).hexdigest(),
|
||||
"job_id": item["job_id"],
|
||||
"file_type": file_type,
|
||||
"consolidation_id": None,
|
||||
}
|
||||
assert len(storage.objects) == len(accepted)
|
||||
|
||||
@ -578,26 +579,33 @@ async def _version_state(session: AsyncSession, version_id: uuid.UUID) -> Consol
|
||||
return (await session.execute(stmt.execution_options(populate_existing=True))).scalar_one()
|
||||
|
||||
|
||||
async def _post_in_version(
|
||||
client: httpx.AsyncClient, entity_id: uuid.UUID, version_id: uuid.UUID, *uploads: Upload
|
||||
async def _post_in_consolidation(
|
||||
client: httpx.AsyncClient, consolidation_id: uuid.UUID, entity_id: uuid.UUID, *uploads: Upload
|
||||
) -> httpx.Response:
|
||||
files = [("files", (name, content, "application/octet-stream")) for name, content in uploads]
|
||||
return await client.post(_files_url(entity_id), params={"consolidation_version_id": str(version_id)}, files=files)
|
||||
return await client.post(f"/consolidations/{consolidation_id}/legal-entities/{entity_id}/files", files=files)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_in_version_moves_initiated_to_documents_processing(
|
||||
async def test_upload_in_consolidation_moves_initiated_to_documents_processing(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
user_id = await _user_id(seeded, KA_RF)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("in-version")))
|
||||
response = await _post_in_consolidation(
|
||||
client, version.consolidation_id, member.id, ("osv.xls", _xls("in-version"))
|
||||
)
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert len(response.json()["accepted"]) == 1
|
||||
assert len(storage.objects) == 1
|
||||
file = await seeded.get(StoredFile, uuid.UUID(response.json()["accepted"][0]["file_id"]))
|
||||
assert file is not None
|
||||
assert file.consolidation_id == version.consolidation_id
|
||||
[upload_event] = await _upload_events(seeded)
|
||||
assert upload_event.after_value["consolidation_id"] == str(version.consolidation_id)
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
history = (
|
||||
(
|
||||
@ -627,13 +635,15 @@ async def test_upload_in_version_moves_initiated_to_documents_processing(
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize("status", [ConsolidationStatus.DOCUMENTS_PROCESSING, ConsolidationStatus.DATA_CORRECTION])
|
||||
async def test_upload_in_version_keeps_other_in_work_status(
|
||||
async def test_upload_in_consolidation_keeps_other_in_work_status(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage, status: ConsolidationStatus
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded, status=status)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls(f"keep-{status.value}")))
|
||||
response = await _post_in_consolidation(
|
||||
client, version.consolidation_id, member.id, ("osv.xls", _xls(f"keep-{status.value}"))
|
||||
)
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert await _version_state(seeded, version.id) is status
|
||||
@ -647,7 +657,7 @@ async def test_upload_with_all_files_rejected_does_not_move_version(
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("notes.docx", ZIP + b"x"))
|
||||
response = await _post_in_consolidation(client, version.consolidation_id, member.id, ("notes.docx", ZIP + b"x"))
|
||||
|
||||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||||
assert response.json()["accepted"] == []
|
||||
@ -664,7 +674,9 @@ async def test_upload_for_entity_outside_perimeter_is_422_and_writes_nothing(
|
||||
client = await auth_client(KA_RF)
|
||||
before = await _counts(seeded)
|
||||
|
||||
response = await _post_in_version(client, outsider.id, version.id, ("osv.xls", _xls("outsider")))
|
||||
response = await _post_in_consolidation(
|
||||
client, version.consolidation_id, outsider.id, ("osv.xls", _xls("outsider"))
|
||||
)
|
||||
|
||||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||||
assert response.json()["detail"]["code"] == "ENTITY_NOT_IN_PERIMETER"
|
||||
@ -674,23 +686,23 @@ async def test_upload_for_entity_outside_perimeter_is_422_and_writes_nothing(
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_in_cancelled_version_is_409(
|
||||
async def test_upload_without_in_work_version_is_409(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded, status=ConsolidationStatus.CANCELLED)
|
||||
client = await auth_client(KA_RF)
|
||||
before = await _counts(seeded)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("cancelled")))
|
||||
response = await _post_in_consolidation(client, version.consolidation_id, member.id, ("osv.xls", _xls("cancelled")))
|
||||
|
||||
assert response.status_code == HTTP_CONFLICT, response.text
|
||||
assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"
|
||||
assert response.json()["detail"]["code"] == "CONSOLIDATION_NOT_EDITABLE"
|
||||
assert await _counts(seeded) == before
|
||||
assert storage.objects == {}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_in_foreign_version_is_404_like_missing(
|
||||
async def test_upload_in_foreign_consolidation_is_404_like_missing(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
foreign, _, _ = await _version(seeded, office_name=KRASNODAR)
|
||||
@ -698,8 +710,12 @@ async def test_upload_in_foreign_version_is_404_like_missing(
|
||||
client = await auth_client(KA_RF)
|
||||
before = await _counts(seeded)
|
||||
|
||||
foreign_response = await _post_in_version(client, own_entity.id, foreign.id, ("osv.xls", _xls("foreign-version")))
|
||||
missing_response = await _post_in_version(client, own_entity.id, uuid.uuid4(), ("osv.xls", _xls("missing-version")))
|
||||
foreign_response = await _post_in_consolidation(
|
||||
client, foreign.consolidation_id, own_entity.id, ("osv.xls", _xls("foreign-version"))
|
||||
)
|
||||
missing_response = await _post_in_consolidation(
|
||||
client, uuid.uuid4(), own_entity.id, ("osv.xls", _xls("missing-version"))
|
||||
)
|
||||
|
||||
assert foreign_response.status_code == missing_response.status_code == HTTP_NOT_FOUND, foreign_response.text
|
||||
assert foreign_response.json() == missing_response.json()
|
||||
@ -718,7 +734,9 @@ async def test_upload_does_not_require_version_lock(
|
||||
await make_lock(seeded, version=version, user=holder)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("under-lock")))
|
||||
response = await _post_in_consolidation(
|
||||
client, version.consolidation_id, member.id, ("osv.xls", _xls("under-lock"))
|
||||
)
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
@ -726,17 +744,28 @@ async def test_upload_does_not_require_version_lock(
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_without_version_does_not_touch_versions(
|
||||
async def test_upload_without_consolidation_does_not_touch_versions(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post(client, member.id, ("osv.xls", _xls("no-version")))
|
||||
ignored = await client.post(
|
||||
_files_url(member.id),
|
||||
params={"consolidation_version_id": str(version.id)},
|
||||
files=[("files", ("osv.xls", _xls("ignored-query"), "application/octet-stream"))],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert ignored.status_code == HTTP_CREATED, ignored.text
|
||||
bodies = (response.json(), ignored.json())
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED
|
||||
assert len(storage.objects) == 1
|
||||
assert len(storage.objects) == len(bodies)
|
||||
for body in bodies:
|
||||
file = await seeded.get(StoredFile, uuid.UUID(body["accepted"][0]["file_id"]))
|
||||
assert file is not None
|
||||
assert file.consolidation_id is None
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@ -755,7 +784,9 @@ async def test_upload_keeps_accepted_files_when_version_transition_fails(
|
||||
messages: list[str] = []
|
||||
sink = logger.add(lambda message: messages.append(str(message)), level="ERROR")
|
||||
try:
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("transition-fails")))
|
||||
response = await _post_in_consolidation(
|
||||
client, version.consolidation_id, member.id, ("osv.xls", _xls("transition-fails"))
|
||||
)
|
||||
finally:
|
||||
logger.remove(sink)
|
||||
|
||||
@ -764,4 +795,26 @@ async def test_upload_keeps_accepted_files_when_version_transition_fails(
|
||||
assert (await seeded.get(StoredFile, uuid.UUID(accepted["file_id"]))) is not None
|
||||
assert len(storage.objects) == 1
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED
|
||||
assert any(str(version.id) in message and "version_transition_error" in message for message in messages)
|
||||
assert any(
|
||||
str(version.consolidation_id) in message and "version_transition_error" in message for message in messages
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_moves_only_versions_with_entity(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
"""Две версии консолидации в INITIATED: ЮЛ в периметре только второй — переходит только она (D-4)."""
|
||||
version, entities, initiator = await _version(seeded)
|
||||
consolidation = await seeded.get(Consolidation, version.consolidation_id)
|
||||
assert consolidation is not None
|
||||
other = await make_version(
|
||||
seeded, consolidation=consolidation, version_no=2, members=[entities[2]], initiator=initiator
|
||||
)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_consolidation(client, consolidation.id, entities[2].id, ("osv.xls", _xls("only-v2")))
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED
|
||||
assert await _version_state(seeded, other.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
|
||||
@ -1159,7 +1159,7 @@ async def test_on_documents_mapped_waits_for_member_edit_and_sees_new_member(
|
||||
async def test_on_documents_mapped_waits_for_transition_to_documents_processing(
|
||||
committed_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""R-2/№40, маппинг ∥ загрузка (`after_upload_to_version`): версия `INITIATED` держится транзакцией, которая
|
||||
"""R-2/№40, маппинг ∥ загрузка (`after_upload_to_consolidation`): версия `INITIATED` держится транзакцией, которая
|
||||
переводит её в `DOCUMENTS_PROCESSING` (не зафиксировано). Хук блокирует надмножество версий «в работе» (а не только
|
||||
`DOCUMENTS_PROCESSING`), поэтому встаёт в ожидание строки версии и после commit видит `DOCUMENTS_PROCESSING` —
|
||||
переход в `DATA_CORRECTION` не теряется. Хук, ищущий версии только в `DOCUMENTS_PROCESSING`, снимок `INITIATED`
|
||||
|
||||
@ -49,6 +49,7 @@ EXPECTED_ROUTES: dict[str, set[str]] = {
|
||||
f"{A}/roles/{{code}}/permissions": {"put"},
|
||||
# primary-files, documents-api (Срез 2c)
|
||||
"/api/v1/legal-entities/{legal_entity_id}/files": {"get", "post"},
|
||||
"/api/v1/consolidations/{consolidation_id}/legal-entities/{legal_entity_id}/files": {"get", "post"}, # v5.18
|
||||
"/api/v1/files/{file_id}": {"get", "delete"},
|
||||
"/api/v1/files/{file_id}/content": {"get"},
|
||||
"/api/v1/documents/{document_id}": {"get"},
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user