mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
33 lines
1.2 KiB
Python
33 lines
1.2 KiB
Python
"""Postgres adapter for ``BulkUploadStatusWriter`` (ADR-0013).
|
|
|
|
Flips the ``bulk_address_uploads`` status on the caller's session — so the
|
|
finaliser can mark ``complete`` in the *same* transaction as the property insert
|
|
(atomic finalise), and mark ``failed`` in a fresh session on the error path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from infrastructure.postgres.bulk_address_upload_table import BulkAddressUploadRow
|
|
from repositories.bulk_upload.bulk_upload_status_writer import BulkUploadStatusWriter
|
|
|
|
|
|
class BulkUploadStatusWriterPostgresRepository(BulkUploadStatusWriter):
|
|
def __init__(self, session: Session) -> None:
|
|
self._session = session
|
|
|
|
def set_status(self, task_id: UUID, status: str) -> None:
|
|
row = self._session.exec(
|
|
select(BulkAddressUploadRow).where(
|
|
BulkAddressUploadRow.task_id == task_id
|
|
)
|
|
).first()
|
|
if row is None:
|
|
raise ValueError(f"No bulk_address_uploads row for task_id {task_id}")
|
|
row.status = status
|
|
row.updated_at = datetime.now(timezone.utc)
|
|
self._session.add(row)
|