cadrumo.application.live.expedientes_ports module

Application-owned capabilities and DTOs for live expedientes reads.

The expedientes lifecycle stores an application snapshot and consumes a read-only declaration-register capability. Browser, Sede DTO, and encrypted storage implementations are composed outside this module and translated at that boundary.

class ExpedientesDeclaration(**data)[source]

Bases: BaseModel

Application DTO for one read-only declaration-register row.

Parameters:

data (Any)

modelo: str
ejercicio: FilingYear
period: Period
expediente_id: AeatExpedienteId
estado: str
tipo_solicitud: str | None
observaciones: str | None
presented_at: datetime
justificante_link_text: str | None
archive_link_text: str | None
declaration_copy_link_text: str | None
justificante_cell_index: int
archive_cell_index: int | None
declaration_copy_cell_index: int | None
mode: Literal['read']
class ExpedientesRegisterProtocol(*args, **kwargs)[source]

Bases: Protocol

Reusable authenticated declaration-register read session.

async walk(*, modelo, ejercicio)[source]

Return application-owned rows for one modelo/year query.

Return type:

tuple[ExpedientesDeclaration, ...]

Parameters:

  • modelo (str)

  • ejercicio (int)

class ExpedientesDeclarationReaderProtocol(*args, **kwargs)[source]

Bases: Protocol

Open a translated, read-only declaration-register session.

open_register(session, *, settings)[source]

Yield a register whose adapter DTOs/errors never cross the port.

Return type:

AbstractAsyncContextManager[ExpedientesRegisterProtocol]

Parameters:
class ExpedientesPorts(declaration_reader, snapshot_repository_factory)[source]

Bases: object

Required live-read and bucket-persistence capabilities.

Parameters:
declaration_reader: ExpedientesDeclarationReaderProtocol
snapshot_repository_factory: Callable[[str], SnapshotRepository[Any]]
class ExpedientesPortsFactory(*args, **kwargs)[source]

Bases: Protocol

Construct expedientes capabilities for one profile bucket.