aeat.application.workflow._models module

Strict pydantic v2 records for the composite workflow engine.

Every boundary-crossing type in aeat.application.workflow is defined here as a frozen, strict, extra="forbid" pydantic.BaseModel or as an enum.StrEnum for closed enumerations. WorkflowStep.details is reserved for string-valued diagnostics emitted by workflow diagnostics. Some helpers accept an optional SecureObjectRepository so callers can supply a custom storage backend without going through the runtime default. The WorkflowState record carries a reference to the active-bucket TransactionCatalogueRepository when one is needed downstream.

This module uses WorkflowResult, WorkflowEngine, and UserProfileRecord for workflow persistence and state management. WorkflowEvent and the review-annotation field types embedded on WorkflowState (InvoiceReviewRecord, LedgerReviewRecord) are defined in the shared leaf module aeat.application._workflow_review_models rather than here or in aeat.application.review, because aeat.application.review embeds WorkflowEvent as a field type in turn — a genuine mutual runtime dependency that a shared leaf module resolves without either package importing the other.

See also

WorkflowEngine

Produces WorkflowResult records and advances WorkflowStage values.

WorkflowPurpose

Selects the local FILE or VERIFY policy that controls deadline and preflight treatment.

WorkflowRunRepository

Persists terminal WorkflowResult records in secure storage.

WorkflowStateRepository

Persists the encrypted WorkflowState envelope.

aeat.application.modelo._workflow_gate

Drives calculation revisions through the workflow and persists the resulting run record before verification or local filing state changes.

Import ordering note

The SiteHealthStatus and ModeloDeadline imports are placed after WorkflowState and related state models so that aeat.application.auth._actions (which imports WorkflowState from this partially-initialised module during the browser-adapter import chain) finds those names already present.

class WorkflowStage(*values)[source]

Bases: StrEnum

The read-only stages of the composite workflow, in strict order.

LOADING_PROFILE
COMPUTING_DEADLINES
CHECKING_INBOX
BUILDING_DRAFT
VALIDATING_DRAFT
RUNNING_PREFLIGHT
DONE
ABORTED
class WorkflowPurpose(*values)[source]

Bases: StrEnum

Why the workflow engine is being driven.

The purpose decides whether the filing-window deadline is an abort gate or merely informational context:

  • FILE — the end-to-end filing pipeline (work file and the end-to-end WorkflowEngine run). Filing without a pending obligation is refused: the COMPUTING_DEADLINES stage aborts with WorkflowAbortReason.NO_PENDING_OBLIGATION when the schedule carries no matching obligation and with WorkflowAbortReason.DEADLINE_PASSED when the obligation window has already closed.

  • VERIFY — the work verify calculation check. Verification asserts a calculation is internally sound against the registry’s verification expectations; it has no honest dependency on the AEAT filing calendar. The COMPUTING_DEADLINES stage records the filing-window state as informational context and never aborts on it, so a correct calculation can be confirmed early, offline, or for a past period.

FILE
VERIFY
class WorkflowAbortReason(*values)[source]

Bases: StrEnum

Closed set of reasons the WorkflowEngine may abort a run.

Each member maps to a distinct failure path in the engine’s stage sequence. CLI surfaces and audit logs carry the string value so operators and tools can key on it without importing this module.

NO_PENDING_OBLIGATION
INBOX_BLOCKING_REQUERIMIENTO
DEADLINE_PASSED
ALREADY_FILED
DRAFT_HAS_ERRORS
PREFLIGHT_FAILED
CERT_INVALID
USER_CANCELLED
SITE_UNAVAILABLE
UNHANDLED_EXCEPTION
class DeclaracionPointer(**data)[source]

Bases: BaseModel

Lightweight pointer to a persisted filing draft stored in WorkflowState.

Keyed in WorkflowState.declarations by the value returned from declaration_key(). draft_id and status are written by the workflow engine after each filing stage; exported_path records the on-disk fichero-BOE path when the draft was exported; verified records the last verification verdict for the work verify command.

Parameters:
modelo: str
period: Period
draft_id: str | None
status: str | None
exported_path: str | None
verified: bool | None
updated_at: datetime
class ProfileBucketPointer(**data)[source]

Bases: BaseModel

Pointer to a secure profile bucket.

bucket_id is the immutable UUIDv4 profile identity and the name of the bucket directory on disk. label is the decoupled mutable operator-chosen display name read from the bucket manifest. status is the plaintext lifecycle marker carried on the manifest; the live-surface scanners filter on it so a tombstoned profile never leaks into list / switch / name-uniqueness.

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

  • label (str)

  • status (BucketLifecycleStatus)

bucket_id: BucketId
label: str
status: BucketLifecycleStatus
declaration_key(modelo, period)[source]

Return the canonical state-store key for a (modelo, period) pair.

The period segment is stored as filing_year:registry_token so declaration state never keys by a combined token such as 2025Q1.

Return type:

str

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

Bases: BaseModel

Encrypted operator state for the AEAT user CLI.

The entire state is persisted as a single encrypted envelope via WorkflowStateRepository. Mutations always return a new copy (model_copy()) to preserve the frozen-model invariant.

Variables:

  • auth – Local AEAT access readiness state.

  • declarations – Filing draft pointers keyed by declaration_key().

  • invoice_reviews – Invoice review annotations keyed by invoice_id.

  • ledger_reviews – Ledger transaction review annotations keyed by transaction_id.

  • updated_at – UTC timestamp of the last write.

Parameters:

The historical profiles field has retired. Consumers that need to enumerate registered profiles call aeat.application.workflow._profile_bucket_scan.list_profile_buckets() or read_profile_bucket() directly; both scan <aeat_local_storage_root>/buckets/*/manifest.toml and never open an encrypted database. The active profile resolves via the precedence chain (Settings override > plaintext pointer file).

auth: AuthState
declarations: dict[str, DeclaracionPointer]
invoice_reviews: dict[str, InvoiceReviewRecord]
ledger_reviews: dict[str, LedgerReviewRecord]
bucket_events: tuple[WorkflowEvent, ...]
updated_at: datetime
active_profile_record(*, secure_objects=None, schema=None)[source]

Return the active UserProfileRecord from its secure bucket.

The active bucket id resolves via the precedence chain in aeat.core.resolve_active_bucket_id() (env var > pointer file fallback). The bucket id and profile name are 1:1 by orchestration convention, so the resolved id is the lifecycle-service read key.

secure_objects (a SecureObjectRepository override) and schema are optional overrides forwarded to build_lifecycle_service(); a per-bucket store and the bundled schema are resolved when None.

Return type:

UserProfileRecord | None

Parameters:
active_profile_bucket_id()[source]

Return the active profile’s secure bucket id via the precedence chain.

Return type:

str | None

active_transaction_catalogue_repository(state, *, objects=None)[source]

Return the TransactionCatalogueRepository for the active profile bucket.

Parameters:
Return type:

TransactionCatalogueRepository

update_declaration_pointer(state, *, modelo, period, draft_id=None, status=None, exported_path=None, verified=None)[source]

Return state with the declaration pointer upserted for (modelo, period).

draft_id and status are optional: when omitted (None) on an update they leave the existing pointer’s value untouched rather than clobbering it, so a partial update (e.g. recording only an exported_path) is safe.

Returns the updated WorkflowState with the pointer recorded.

Return type:

WorkflowState

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

Bases: BaseModel

Workflow-side alert wrapping a SiteHealthStatus observation.

Attached to a WorkflowStep when the AEAT browser health-check adapter reports a non-nominal site status during a workflow run. stage identifies the workflow stage that observed the alert; run_id ties it to the enclosing WorkflowResult.

Parameters:
stage: WorkflowStage
status: SiteHealthStatus
run_id: str
class WorkflowStepDetails(**data)[source]

Bases: BaseModel

Operator-visible details attached to a workflow step.

Carries arbitrary string-keyed diagnostic values emitted by the workflow engine. The model is intentionally permissive (extra='allow') so existing call sites can continue to pass free-form dicts; the boundary is now a typed pydantic record instead of an opaque dict[str, str], so a future PR can promote specific step kinds into a discriminated union without breaking the field type on WorkflowStep.

Per-stage key catalogue (the documented contract external tools consuming WorkflowResult.steps may rely on):

  • Deadline checks: {"modelo", "period", "closes_on"}.

  • Draft / snapshot mismatch: {"draft_id", "modelo", "period", "profile_tax_id", "schema_version"}.

  • Calculation validation failure: {"error_count"} and any issue-specific keys.

  • AEAT certificate health: {"provider_kind", "provider_operator_impact", "cert_not_after", "cert_severity", "cert_days_until_expiry"}.

  • Site-health alerts: {"status", "run_id"}.

New keys may be added by the engine without bumping the workflow schema version. Removal or rename is a breaking change and must be paired with a workflow schema-version bump.

Implements __getitem__, __contains__, and get so existing read-side code that treats step.details like a Mapping[str, str] keeps working without per-call-site migration; the typed model now anchors the storage shape.

Frozen and strict on the inner values to preserve the boundary-strictness guarantee on workflow diagnostics.

Parameters:

extra_data (Any)

get(key, default=None)[source]

Return the diagnostic value for key, or default if absent.

Return type:

object

Parameters:
items()[source]

Return all extra diagnostic key-value pairs as a Mapping.

Return type:

Mapping[str, object]

class WorkflowStep(**data)[source]

Bases: BaseModel

A single step in a WorkflowResult.

Parameters:
stage: WorkflowStage
started_at: datetime
ended_at: datetime | None
success: bool | None
summary: str
details: Annotated[WorkflowStepDetails | Mapping[str, object] | None, BeforeValidator(_coerce_workflow_step_details)]
site_health_alert: SiteHealthAlert | None
class WorkflowResult(**data)[source]

Bases: BaseModel

The full result of one WorkflowEngine.run_next() invocation.

Parameters:
run_id: str
started_at: datetime
ended_at: datetime
final_stage: WorkflowStage
aborted_reason: WorkflowAbortReason | None
obligation: ModeloDeadline | None
draft_id: str | None
submission_id: str | None
steps: tuple[WorkflowStep, ...]
summary: str
resumed_from: str | None
compute_run_id(*, tax_id, modelo, period, started_at)[source]

Return a stable 16-char hex hash for a workflow run.

Return type:

str

Parameters: