aeat.domain.modelos._work_unit module

Modelo work-unit value records.

WorkUnit and WorkUnitCatalogue bind a ModeloCode, Period, BucketId, and WorkUnitId into the operator-facing handle addressed by derive_work_unit_id().

A modelo work unit is a stable, bucket-scoped handle over one calculation revision of a specific (modelo, year, period). It is the operator-facing object the modelo workflow verbs (calculate, verify, file, filing-record, etc.) attach state to. The work unit itself carries metadata only — its calculation results, verification reports, and filing records live in separate stores keyed by work_unit_id.

The work-unit identifier is a SHA-256 hex digest derived deterministically from the four-axis key (bucket_id, modelo, filing_year, period, revision_id). The deterministic derivation means two consumers of the same four-axis key see the same work_unit_id without round-tripping through storage. Renaming a work unit does not change the identifier; name is a display-only attribute.

class WorkUnitState(*values)[source]

Bases: StrEnum

Closed enumeration of work-unit lifecycle states.

  • BORRADOR — default state at creation. The work unit participates in default listings and accepts mutation (rename, future calculation revisions).

  • DESCARTADO — operator marked the work unit abandoned. Excluded from default listings; mutations are rejected. Revision payloads remain in storage for audit; the work unit cannot be re-activated.

BORRADOR
DESCARTADO
derive_work_unit_id(*, bucket_id, modelo, filing_year, period, revision_id)[source]

Return the deterministic work_unit_id for a four-axis key.

The five inputs are normalised (stripped, lowercased where case is insignificant) and joined with a stable separator that cannot appear inside any of the inputs. The resulting bytes are hashed with SHA-256 and rendered as 64-character lowercase hex — identical to the catalogue-key shape the project uses elsewhere for content-addressed identifiers.

Determinism is the operative contract: callers that build the same four-axis key see the same identifier without round-tripping through storage.

Return type:

str

Parameters:
class WorkUnit(**data)[source]

Bases: BaseModel

One operator-facing modelo calculation work unit.

Variables:

  • work_unit_id – Lowercase 64-char SHA-256 derived from (bucket_id, modelo, filing_year, period, revision_id) by derive_work_unit_id(). The identifier is stable across renames.

  • bucket_id – Stable bucket identity the work unit lives inside. Two work units with the same modelo / year / period / revision in different buckets have different work_unit_id values.

  • modelo – AEAT modelo code (e.g. "303", "130").

  • filing_year – Tax year for which the modelo is being filed.

  • period – Typed filing period value carrying the filing year and bare registry token (e.g. "4T", "0A", "01").

  • revision_id – Stable id of the registry-known modelo revision the work unit targets.

  • name – Display name. Defaults to "<modelo>-<year>-<period>" when the operator does not supply one explicitly.

  • created_at – Timezone-aware UTC timestamp at first creation.

  • updated_at – Timezone-aware UTC timestamp at the most recent mutation. Equals created_at on a fresh work unit.

  • state – Lifecycle state — BORRADOR by default, DESCARTADO once the operator marks the unit abandoned via the discard verb.

  • discarded_at – Timezone-aware UTC timestamp set at discard time. None for non-discarded units.

  • discarded_by – Actor label captured at discard time. None for non-discarded units.

  • discard_reason – Operator-supplied free-text reason for the discard. None when no reason was given (or when the unit is not discarded).

  • censo_stamped_stale_at – Timezone-aware UTC timestamp the stale-cascade walker set when aeat config profile censo apply superseded the censo facts the work unit depended on. None while the unit is still censo-current. Set/unset together with censo_stale_reason.

  • censo_stale_reason – Operator-readable text recording why the unit was marked stale (typically the superseding snapshot id). None while not stale; both this and censo_stamped_stale_at set together when stale, both None otherwise.

Parameters:

  • 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)])

  • 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)])

  • modelo (ModeloCode)

  • filing_year (Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=2000), Le(le=2099)])])

  • period (Period)

  • revision_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)])

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

  • created_at (datetime)

  • updated_at (datetime)

  • state (WorkUnitState)

  • discarded_at (datetime | None)

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

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

  • current_calculation_revision_id (Annotated[str | None, 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)])

  • filed_calculation_revision_id (Annotated[str | None, 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)])

  • current_filing_record_id (Annotated[str | None, 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)])

  • censo_stamped_stale_at (datetime | None)

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

  • ledger_stamped_stale_at (datetime | None)

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

  • causante_ccaa (CCAA | None)

work_unit_id: WorkUnitId
bucket_id: BucketId
modelo: ModeloCode
filing_year: Annotated[int, Field(ge=2000, le=2099)]
period: Period
revision_id: _RevisionId
name: _DisplayName
created_at: datetime
updated_at: datetime
state: WorkUnitState
discarded_at: datetime | None
discarded_by: ModeloActorLabel | None
discard_reason: _DiscardReason | None
current_calculation_revision_id: _OptionalHex64
filed_calculation_revision_id: _OptionalHex64
current_filing_record_id: _OptionalHex64
censo_stamped_stale_at: datetime | None
censo_stale_reason: _StaleReason | None
ledger_stamped_stale_at: datetime | None
ledger_stale_reason: _StaleReason | None
causante_ccaa: CCAA | None
class WorkUnitCatalogue(**data)[source]

Bases: BaseModel

Immutable catalogue of every work unit known in storage.

The catalogue is a frozen mapping keyed by work_unit_id. Lookups by other shapes (bucket, modelo, year, period) are iterated at the catalogue boundary; the catalogue itself does not index secondary keys.

Parameters:

work_units (Mapping[str, WorkUnit])

work_units: Mapping[str, WorkUnit]
classmethod from_work_units(units)[source]

Build a WorkUnitCatalogue from an iterable / mapping of work units.

Return type:

WorkUnitCatalogue

Parameters:

units (Mapping[str, WorkUnit] | tuple[WorkUnit, ...])

get(work_unit_id)[source]

Return the work unit for work_unit_id or None if absent.

Return type:

WorkUnit | None

Returns:

The WorkUnit for the given id, or None when not found.

Parameters:

work_unit_id (str)

values()[source]

Return a view of every WorkUnit in the catalogue.

Return type:

ValuesView[WorkUnit]