aeat.application.modelo._history module

Per-work-unit chronological history assembly.

Surfaces a unified timeline of every event the operator’s work unit emitted across its lifecycle: creation, renames, calculations, verifications, filings, supersessions, amendments, and discards.

The catalogue substrate is the bucket-scoped append-only event log loaded from BucketEventHistoryRepository. Events scoped to a work unit land under four aeat.domain.buckets.BucketEventObjectType values: WORK_UNIT, CALCULATION_REVISION, VERIFICATION_REPORT, and FILING_RECORD. The assembler walks each related object id and merges the emitted aeat.domain.buckets.BucketEvent streams in chronological order.

The normalized records remain the source of relational truth: aeat.domain.modelos.WorkUnit selects the lifecycle root, CalculationRevision identifies calculation attempts under it, aeat.domain.modelos.VerificationReport identifies verification outcomes for those revisions, and ModeloRecord identifies local filing records. The event history explains how those records changed; it does not replace their catalogues.

The assembler is pure read: no mutation, no remote contact.

See also

aeat.application.modelo.create_work_unit():

Emits work-unit create, rename, and discard events.

aeat.application.modelo.calculate_modelo_work_revision():

Persists calculation revisions and MODELO_CALCULATION_CREATED events.

aeat.application.modelo.verify_modelo_revision():

Persists verification reports and verification pass/refusal events.

aeat.application.modelo.file_modelo_revision():

Persists local filing records and filing/supersession events.

class WorkUnitHistoryEvent(**data)[source]

Bases: BaseModel

One projected aeat.domain.buckets.BucketEvent row in a work-unit history stream.

Parameters:
event_id: str
occurred_at: datetime
event_type: BucketEventType
object_type: BucketEventObjectType
object_id: str
actor: str
payload: dict[str, str]
class WorkUnitHistory(**data)[source]

Bases: BaseModel

Chronologically ordered event timeline for one aeat.domain.modelos.WorkUnit.

Parameters:

  • bucket_id (Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=128, pattern=None, ascii_only=None)])

  • work_unit_id (Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=64, max_length=64, pattern=^[0-9a-f]{64}$, ascii_only=None)])

  • events (tuple[WorkUnitHistoryEvent, ...])

bucket_id: BucketId
work_unit_id: WorkUnitId
events: tuple[WorkUnitHistoryEvent, ...]
assemble_work_unit_history(work_unit_id, *, work_unit_repository=None, calculation_repository=None, filing_repository=None, verification_repository=None, bucket_event_repository=None)[source]

Return a WorkUnitHistory covering every bucket event scoped to work_unit_id.

Events are merged from object-scoped aeat.domain.buckets.BucketEvent streams and ordered by occurred_at ascending. The work unit itself is loaded to confirm it exists (raising WorkUnitNotFoundError if not) and to discover every CalculationRevision, aeat.domain.modelos.VerificationReport, and ModeloRecord id that belongs to its lifecycle.

The returned WorkUnitHistoryEvent rows copy event payloads into a read model for aeat app modelo work history. The function never writes to repositories and never contacts AEAT; mutation and event emission stay with the lifecycle services that produce the underlying records.

See also

aeat.domain.buckets.BucketEventHistoryCatalogue.for_object():

Supplies each object-scoped event stream merged here.

WorkUnitHistory:

The immutable read model returned to callers.

Return type:

WorkUnitHistory

Parameters: