cadrumo.application.live.notification_ports module¶

Application-owned ports and DTOs for live notifications.

The live notification service consumes these structural ports. Concrete sede, PDF/parser and persistence adapters are assembled by an entrypoint instead of being imported by the application service.

class NotificationType(*values)[source]¶

Bases: StrEnum

Application-owned classification of one remote notification row.

NOTIFICACION¶

COMUNICACION¶

PENDIENTE¶

UNKNOWN¶

class RemoteNotification(**data)[source]¶

Bases: BaseModel

Application DTO for one observed AEAT notification row.

The Sede adapter owns its scraped DTO and translates it to this record before the row enters application orchestration or encrypted custody.

Parameters:: data (Any)

certificado_id: AeatCertificadoId¶

tipo: NotificationTypeValue¶

concepto: str¶

titular_nif: str¶

titular_nombre: str¶

destinatario_nif: str¶

destinatario_nombre: str¶

fecha_emision: date¶

fecha_notificacion: date | None¶

modo_notificacion: str | None¶

leida: bool | None¶

source_url: str¶

mode: Literal['read']¶

class NotificationsSnapshot(**data)[source]¶

Bases: BaseModel

Application DTO for one read-only notifications query result.

Parameters:: data (Any)

rows: tuple[RemoteNotification, ...]¶

captured_at: datetime¶

source_url: str¶

mode: Literal['read']¶

class NotificationSnapshotQueryProtocol(*args, **kwargs)[source]¶

Bases: Protocol

Read the authenticated notification surface and translate its result.

async fetch(session, *, settings)[source]¶

Return one application-owned notifications snapshot.

Return type:

NotificationsSnapshot

Parameters:

session (AeatSession)
settings (Settings)

class NotificationsPorts(snapshot_query, snapshot_repository_factory)[source]¶

Bases: object

Required live-query and bucket-persistence capabilities for notifications.

Parameters:

snapshot_query (NotificationSnapshotQueryProtocol)
snapshot_repository_factory (Callable[[str], SnapshotRepository[Any]])

snapshot_query: NotificationSnapshotQueryProtocol¶

snapshot_repository_factory: Callable[[str], SnapshotRepository[Any]]¶

class NotificationRowProtocol(*args, **kwargs)[source]¶

Bases: Protocol

The notification-row facts needed by the custody use case.

property certificado_id: StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=10, max_length=16, pattern=^\d{10,16}$, ascii_only=None)]¶: Execute this public contract operation.

property tipo: str¶: Execute this public contract operation.

property fecha_emision: date¶: Execute this public contract operation.

property leida: bool | None¶: Execute this public contract operation.

class NotificationDocumentProtocol(*args, **kwargs)[source]¶

Bases: Protocol

The in-memory document facts needed by the custody use case.

property certificado_id: StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=10, max_length=16, pattern=^\d{10,16}$, ascii_only=None)]¶: Execute this public contract operation.

property pdf_bytes: bytes¶: Execute this public contract operation.

property pdf_sha256: str¶: Execute this public contract operation.

property source_url: object¶: Execute this public contract operation.

class NotificationDocumentReaderProtocol(*args, **kwargs)[source]¶

Bases: Protocol

Read document bytes into a domain reading or an explicit refusal.

read(document)[source]¶

Execute this public contract operation.

Return type:: tuple[SancionLiquidacion | None, str | None]
Parameters:: document (NotificationDocumentProtocol)