cadrumo.application.auth.operator_scope_ports module

Application-owned ports for operator auth storage scoping.

Operator auth policy needs only a narrow bucket-path/lock surface and a read-only observation of the currently bound custody session. The concrete filesystem paths, lock records, and live session objects remain owned by the outward persistence adapters; this module carries the translated records and the capability bundle that application callers require.

class OperatorScopeBucketPaths(bucket_id, bucket_dir)[source]

Bases: object

The path facts required to lock one operator-auth bucket.

Parameters:

  • bucket_id (str)

  • bucket_dir (Path)

bucket_id: str
bucket_dir: Path
class OperatorScopeSession(bucket_id, storage_root)[source]

Bases: object

The non-secret identity facts of the currently bound custody session.

Parameters:

  • bucket_id (str)

  • storage_root (Path | None)

bucket_id: str
storage_root: Path | None
class OperatorScopeStoragePort(*args, **kwargs)[source]

Bases: Protocol

Resolve and lock a bucket without exposing persistence DTOs.

resolve(root, bucket_id)[source]

Resolve the bucket directory used by the auth mutation lock.

Return type:

OperatorScopeBucketPaths

Parameters:
acquire_lock(paths, *, wait_seconds)[source]

Acquire the bucket lock or raise an application storage error.

Return type:

None

Parameters:
release_lock(paths)[source]

Release the bucket lock owned by the current auth mutation.

Return type:

None

Parameters:

paths (OperatorScopeBucketPaths)

class OperatorScopeSessionPort(*args, **kwargs)[source]

Bases: Protocol

Observe and compare the live custody session for one auth operation.

current()[source]

Return the translated current session, if one is bound.

Return type:

OperatorScopeSession | None

serves_bucket(session, bucket_id)[source]

Return whether the supplied session serves the exact bucket.

Return type:

bool

Parameters:
class OperatorScopePorts(storage, session)[source]

Bases: object

Required outward capabilities for operator auth storage scope policy.

Parameters:
storage: OperatorScopeStoragePort
session: OperatorScopeSessionPort
exception OperatorScopeStorageError(*, operation, bucket_id, failure_kind)[source]

Bases: CadrumoError

Translated failure from bucket path or lock persistence.

Parameters:

  • operation (str)

  • bucket_id (str)

  • failure_kind (str)

Return type:

None

code
exception OperatorScopeSessionError(message=None, *, context=None, translated_message=None)[source]

Bases: CadrumoError

Translated failure while observing the active custody session.

Parameters:

  • message (str | None)

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

  • translated_message (str | None)

Return type:

None

code