aeat.application.auth._acquisition_lock module

Crash-recoverable auth acquisition locks.

The lock protects live auth flows that can create external state, especially Cl@ve Movil push petitions. It is intentionally filesystem-backed so separate CLI processes share the same guard. The lock file stores an AuthAcquisitionLockRecord and reports operator-safe state through AuthAcquisitionLockStatus.

class AuthAcquisitionLockState(*values)[source]

Bases: StrEnum

Observable states for the auth acquisition lock file.

ABSENT
HELD
STALE
CORRUPT
class AuthAcquisitionLockRecord(**data)[source]

Bases: BaseModel

Metadata written into an auth acquisition lock file.

Parameters:
provider_kind: AuthProviderKind
profile_name: str
pid: int
hostname: str
created_at: datetime
expires_at: datetime
operation: str
class AuthAcquisitionLockStatus(**data)[source]

Bases: BaseModel

Safe health/status view of an auth acquisition lock.

Parameters:
state: AuthAcquisitionLockState
path: Path
record: AuthAcquisitionLockRecord | None
reason: str | None
recoverable: bool
property locked: bool

Return True when another live process should block auth acquisition.

exception AuthAcquisitionLockedError(message=None, *, context=None, suggestion=None, translated_message=None)[source]

Bases: AeatError

Raised when another process is already acquiring AEAT auth.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • suggestion (str | None)

  • translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]
auth_acquisition_lock_path(settings, kind)[source]

Return the profile/provider-scoped lock path.

Return type:

Path

Parameters:
inspect_auth_acquisition_lock(settings, kind, *, now=None)[source]

Describe the current acquisition-lock health without mutating it.

Returns an AuthAcquisitionLockStatus.

Return type:

AuthAcquisitionLockStatus

Parameters:
clear_auth_acquisition_lock(settings, kind, *, reason='operator-reset')[source]

Remove the acquisition lock and return the pre-reset status.

Returns an AuthAcquisitionLockStatus reflecting the state observed immediately before the file was removed.

Return type:

AuthAcquisitionLockStatus

Parameters:
acquire_auth_acquisition_lock(settings, kind, *, ttl_seconds, operation='auth-login')[source]

Acquire a crash-recoverable auth lock or raise a typed conflict.

Yields an AuthAcquisitionLockRecord while the lock is held.

Stale/corrupt locks are removed automatically before a second atomic-create attempt. A live lock is never waited on or retried: callers fail early so they do not issue a duplicate Cl@ve petition.

Return type:

Iterator[AuthAcquisitionLockRecord]

Parameters:
auth_lock_ttl_seconds(settings, kind)[source]

Return the acquisition-lock TTL for a provider.

Return type:

int

Parameters: