aeat.application.workflow._persistence module¶

Encrypted persistence for workflow state and workflow runs.

Workflow state is stored as an Envelope-wrapped record in the secure-object backend. The load path deserialises the envelope and validates it; callers receive a typed WorkflowState or a diagnostic error class rather than a raw payload.

See also

WorkflowState: Typed encrypted state payload persisted by WorkflowStateRepository.
WorkflowStateResetFingerprint: Row-level, plaintext-free reset audit summary emitted before deletion.
application.workflow._events.emit_workflow_state_reset(): Writes the append-only workflow_state.reset bucket event before the state row is removed.
BucketEventHistoryRepository: Stores the emitted reset event in the bucket event history.
WorkflowResult: Terminal workflow run record persisted separately by WorkflowRunRepository.

class WorkflowEnvelopeReasonClass(*values)[source]¶

Bases: StrEnum

Classification of the workflow-state envelope’s readability.

Carried in reason_class to distinguish a healthy envelope (READABLE), a row that cannot be decrypted (UNREADABLE), and an absent row (ABSENT).

READABLE¶

UNREADABLE¶

ABSENT¶

class WorkflowStateRepository(*, objects=None, emit_reset=<function emit_workflow_state_reset>)[source]¶

Bases: object

Encrypted secure-object repository for WorkflowState.

Parameters:

objects (SecureObjectRepository | None)
emit_reset (Callable[..., object])

load()[source]¶

Load state or return an empty payload when absent.

Returns the persisted WorkflowState, or an empty default when no state has been saved yet.

Return type:: WorkflowState

save(state)[source]¶

Persist state in the encrypted database object store.

Return type:: None
Parameters:: state (WorkflowState)

to_secure_object_write(state)[source]¶

Return the secure-object upsert for state without committing it.

Lets callers co-transactionally persist the workflow state and a sibling secure-object payload (typically an updated bucket-event-history catalogue) via a single save_many() call.

Parameters:: state (WorkflowState)

fingerprint_state(*, reason_class=None)[source]¶

Return a WorkflowStateResetFingerprint of the persisted state envelope.

Reads row-level metadata only; never decrypts the payload for the fingerprint fields. The state envelope is loaded once to derive recovered_bucket_id and to classify the envelope’s readability — a healthy, decryptable envelope is reported with reason_class="readable", an absent envelope with "absent", and an envelope row that cannot be decoded with "unreadable". A freshly-created storage root that has only just persisted a healthy state must therefore report readable, never unreadable.

reason_class may be supplied to override the derived classification when the caller already knows the trigger that forced the reset (e.g. a downstream handler that caught the concrete failure). When None the classification is derived from the envelope itself.

The repair reset-progress recovery verb is bootstrap-exempt and may run on a cold root where aeat_database_url does not resolve (no active profile). In that case there is no state envelope to reset; the fingerprint records empty metadata rather than crashing on the absent database (disaster ADR Ruling 6).

Return type:: WorkflowStateResetFingerprint
Parameters:: reason_class (str | None)

reset_workflow_state(*, actor='aeat.application.workflow', source='aeat config repair reset-progress', reason_class=None)[source]¶

Delete the workflow-state envelope and emit a reset event.

The mutation is scoped to namespace aeat.workflow / key state; no other namespace or row is touched. The workflow_state.reset bucket event is appended BEFORE the secure-object row is deleted so the worst-case failure mode leaves an audit entry with the data still present (an idempotent recoverable state) rather than the data discarded without a trail. The fingerprint never carries plaintext envelope content.

Returns a WorkflowStateResetFingerprint with a hash of the deleted state for audit traceability.

Return type:

WorkflowStateResetFingerprint

Parameters:

actor (str)
source (str)
reason_class (str | None)

update(fn)[source]¶

Load, transform, save, and return the updated WorkflowState.

Return type:: WorkflowState
Parameters:: fn (Callable[[WorkflowState], WorkflowState])

class WorkflowRunRepository(*, objects=None)[source]¶

Bases: object

Encrypted secure-object repository for WorkflowResult runs.

Parameters:: objects (SecureObjectRepository | None)

save(result, *, runs_dir=None)[source]¶

Persist one workflow result in the secure object backend.

Return type:

Path

Parameters:

result (WorkflowResult)
runs_dir (Path | None)

load(run_id)[source]¶

Load one persisted WorkflowResult from the secure backend.

Returns the WorkflowResult for run_id.

Return type:: WorkflowResult
Parameters:: run_id (str)

list(*, since=None)[source]¶

List persisted workflow runs newest-first, optionally filtered by date.

Each element is a WorkflowResult.

Return type:: tuple[WorkflowResult, ...]
Parameters:: since (date | None)

workflow_state_repository()[source]¶

Return the WorkflowStateRepository bound to the active-bucket database.

When an active profile bucket is present, the repository is backed by the bucket’s own encrypted database resolved through secure_object_repository_for_active_bucket() so the URL is derived from the live bucket path rather than the settings-override snapshot captured at test-fixture construction time. A cold root with no active bucket pointer is the bootstrap exception: it receives an explicit bare SecureObjectRepository so bootstrap-exempt recovery reads can still observe an absent state.

Return type:: WorkflowStateRepository

reset_workflow_state(*, actor='aeat.application.workflow', source='aeat config repair reset-progress', reason_class=None)[source]¶

Module-level helper around WorkflowStateRepository.reset_workflow_state().

Returns a WorkflowStateResetFingerprint with a hash of the deleted state for audit traceability.

Return type:

WorkflowStateResetFingerprint

Parameters:

actor (str)
source (str)
reason_class (str | None)

fingerprint_workflow_state(*, reason_class=None)[source]¶

Return a WorkflowStateResetFingerprint via WorkflowStateRepository.fingerprint_state().

Return type:: WorkflowStateResetFingerprint
Parameters:: reason_class (str | None)

save_run(result, *, runs_dir=None)[source]¶

Persist one workflow result in the secure object backend.

runs_dir remains part of the API as a logical marker path for callers and tests, but no plaintext run file is written there.

Return type:

Path

Parameters:

result (WorkflowResult)
runs_dir (Path | None)

load_run(run_id)[source]¶

Load and return one WorkflowResult from the secure backend.

Return type:: WorkflowResult
Parameters:: run_id (str)

list_runs(*, since=None)[source]¶

List persisted WorkflowResult runs newest-first, optionally filtered by date.

Return type:: tuple[WorkflowResult, ...]
Parameters:: since (date | None)