`sandloom.sdk.authz`¶

Authorization: policies and atoms, access facts, the billing-account subject (Account).

Public API 1.0 (spec §2.2, Appendix C).

from sandloom.sdk.authz import <name>. Signatures in a terminal: sandloom docs <name>. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line # sandloom: advanced <reason> in generated code and examples).

Start with (beginner)¶

Name	Kind	Summary
`authorize_resource`	function	Load a resource by path id, authorize, and return it to the endpoint.
`Credits`	class	Subject's live credit balance is at least `minimum`.
`OrgRole`	class	Subject holds at least `minimum` role in the request's organization.
`Owner`	class	Subject owns the request resource.
`require`	function	Return a dependency enforcing `policy` (no request resource).
`Subscribed`	class	Subject's subscription grants access (coarse, claim-based).
`SubscriptionActive`	class	Subject's subscription is active right now (live, authoritative).

Advanced¶

Name	Kind	Summary
`AccessDecision`	class	The outcome of evaluating a policy against an `AuthzContext`.
`AccessDenied`	exception	Raised by `enforce` (and the HTTP guards) when a policy is not satisfied.
`AccessFact`	class	Declarative binding of a fact name to its three resolution paths.
`Account`	class	The billing account an atom checks (`of=`); build it with `user`, `org` or `bound_tenant`.
`ACTIVE_SUBSCRIPTION_STATUSES`	constant
`And`	class	Conjunction: all operands must hold. Short-circuits on first non-allow.
`authorize`	async function	Inline check over an already-loaded `resource`; raises on failure.
`AuthzContext`	class	Inputs available to a `Requirement` during evaluation.
`Category`	class	Mandatory denial categories (TZ §10.2).
`ClaimRule`	class	Declarative claim -> bool interpretation: wire-safe, no callables.
`current_context`	async function	Authenticated subject from the Bearer access token (cached per request).
`enforce`	async function	Evaluate `policy` for `subject`; raise `AccessDenied` on deny.
`Entitled`	class	Subject satisfies the registered access fact `fact_name`.
`ENTITLEMENTS`	constant
`EntitlementsAdapter`	class	Answers subscription / credit questions (`SubscriptionActive`/`Credits`).
`FactDenied`	exception	A resolver's definite denial for a reason other than "the fact is False".
`FactOwnerUnreachable`	exception	The fact's owner is bound but could not describe its facts (503).
`FactUnavailable`	exception	A fact could not be resolved in any acceptable mode (a misconfiguration).
`Feature`	class	The account's plan enables the feature flag `key` (`of=`, §7.2).
`lookup_access_fact`	async function	The fact `name`: the local registration, else the owner's public one.
`make_rebuild_gate`	function	A startup task that rebuilds an authz projection from its sources.
`MEMBERSHIP`	constant
`MembershipAdapter`	class	Answers organization-role questions for `OrgRole`.
`MODE_AUTO`	constant
`MODE_CLAIM`	constant
`MODE_LIVE`	constant
`MODE_PROJECTION`	constant
`Not`	class	Negation of a single requirement.
`Or`	class	Disjunction: any operand may hold. Short-circuits on first allow.
`Perm`	class	Subject holds the fine-grained permission `perm`.
`Permissions`	class	Predefined fine-grained permission codes.
`PolicyIndeterminate`	exception	Raised by an atom's `evaluate` when it cannot answer at all.
`register_authz_projection`	function	Register an authorization projection, its consumers and rebuilder.
`register_live_adapter`	function	Bind the adapter the live atoms use for `kind` — CONSUMER-side config.
`Requirement`	class	An authorization predicate, composable with `& \\| ~`.
`resolve_fact`	async function	Resolve `fact` to a boolean, caching the result on `ctx`.
`Role`	class	Subject holds the global role `role`.
`Roles`	class	Predefined global role codes.
`service_subject`	function	A subject representing another service calling across a module boundary.
`SOURCE_HTTP`	constant
`SOURCE_INTERNAL`	constant
`SOURCE_SYSTEM`	constant
`SOURCE_TASK`	constant
`system_subject`	function	A subject representing the platform itself (trusted background work).
`TIER_CLAIM`	constant
`TIER_LIVE`	constant
`TIER_RESOURCE`	constant
`WithinLimit`	class	`used` is strictly below the account's plan limit for `key`.

Beginner names¶

`authorize_resource`¶

def authorize_resource(
    *,
    loader: Callable[[Any, AsyncSession], Awaitable[Any]],
    by: str,
    policy: Requirement,
    cast: Callable[[str], Any] | None = None,
    not_found_detail: str = 'Not found',
    status_code: int | None = None,
) -> Callable[..., Awaitable[Any]]

Load a resource by path id, authorize, and return it to the endpoint.

Args:

loader: async (id_value, db) -> entity | None — the module supplies its own loader (keeps shared free of ORM imports).
by: name of the path parameter holding the resource id.
policy: requirement to enforce; resource atoms (Owner) read the loaded entity.
cast: optional converter applied to the raw path value before loader (e.g. UUID). Defaults to passing the raw string.
not_found_detail: 404 message when the loader returns None.
status_code: explicit status override for a definite denial (default: the decision's own status).

The resource is fetched once: claim atoms evaluate first, so a failing cheap check raises without ever loading; otherwise the entity loaded for the ownership check is the same one returned to the endpoint.

`Credits`¶

class Credits(_TenantScopedLiveAtom)(
    minimum: int,
    *,
    of: Account = <object object>,
    adapter: Any = None,
) -> None

Subject's live credit balance is at least minimum.

Reads the real-time balance via the entitlements contract (never a stale token claim — this guards money). of= names the billing account.

`Credits.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Credits.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`OrgRole`¶

class OrgRole(Requirement)(
    minimum: str,
    *,
    org_param: str = 'org_id',
    adapter: Any = None,
) -> None

Subject holds at least minimum role in the request's organization.

Relationship (ReBAC) authorization: org membership/role is per-organization and not carried in the global JWT, so it is resolved live through the membership contract (local call in a monolith, an RPC when tenancy is a separate service; 501 when neither). The organization id is read from the request path parameter org_param, else X-Tenant-Id, else the bound tenant; a missing organization is indeterminate (denies even under Not).

For a hot path, a module may instead maintain an authorization projection of memberships and gate on a projection-backed Entitled fact.

`OrgRole.org_params`¶

def org_params(self) -> frozenset[str]

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, X-Tenant-Id, else the bound tenant): create_app then refuses a route that solves the gate before its TenantScope() (SANDLOOM-B007).

`OrgRole.risks`¶

def risks(self) -> frozenset[str]

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): {"admin"} for an admin role, else none. An Or carries its branches' risks; a Not grants no authority.

`OrgRole.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`OrgRole.adapter`¶

def adapter(self) -> Any

`OrgRole.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`OrgRole.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`Owner`¶

class Owner(Requirement)(get_owner: Callable[[Any], UUID | None]) -> None

Subject owns the request resource.

get_owner maps the loaded resource to its owner id; it is supplied by the module declaring the policy (e.g. lambda post: post.author_id), so the atom never needs to know the resource's type.

`Owner.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Owner.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`require`¶

def require(
    policy: Requirement,
    *,
    status_code: int | None = None,
) -> Callable[..., Awaitable[UserContext]]

Return a dependency enforcing policy (no request resource).

Use in dependencies=[Depends(require(...))] or as a parameter when the endpoint wants the resulting UserContext. A denial answers with the decision's own status (e.g. Subscribed(of=Account.user()) → 402, a missing contract → 501); status_code is an explicit override for definite denials.

A claims-only policy (every atom tier 1) never opens a database session / Unit of Work; only policies that may reach a live atom depend on get_db.

Raises ValueError at wiring time if policy contains a resource-scoped atom (e.g. Owner): require has no request resource, so such a policy belongs in authorize_resource / authorize.

`Subscribed`¶

class Subscribed(Requirement)(
    statuses: frozenset[str] | None = None,
    *,
    of: Account = <object object>,
    claim_key: str = 'subscription_status',
) -> None

Subject's subscription grants access (coarse, claim-based).

of=Account.user(): reads the token's subscription_status claim with ZERO I/O, interpreted by the subscription.active access fact — billing's declarative rule, the same predicate as the live check (active/trialing pass; past_due passes during the grace period unless BILLING_DENY_ON_PAST_DUE). The fact is the local registration or, in a process without billing, the one billing publishes over billing.facts.v1 (described once, cached).

Provenance is honoured: when the token's authoritative provenance shows billing did NOT contribute (it failed at mint, or the issuer could not reach it) the claim proves nothing, so the atom escalates to billing's LIVE answer instead of denying 402 for the token's whole lifetime (503 indeterminate when no live path exists). With no subscription owner in the deployment at all, only active/trialing pass. May be up to the token TTL stale; for real-time / money-sensitive checks use sandloom.shared.authz.atoms_live.SubscriptionActive / Credits.

of=Account.org(...): the token carries NO per-organization subscription claim (claims are personal-only), so this atom resolves LIVE — it delegates to SubscriptionActive(of=...) and is a tier-3 atom. of=Account.bound_tenant() resolves the bound tenant's kind first (personal → the claim, organization → live).

Passing explicit statuses pins the accepted set verbatim (no grace interpretation).

`Subscribed.tenant_type`¶

tenant_type  (property)

`Subscribed.org_params`¶

def org_params(self) -> frozenset[str]

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, X-Tenant-Id, else the bound tenant): create_app then refuses a route that solves the gate before its TenantScope() (SANDLOOM-B007).

`Subscribed.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`Subscribed.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Subscribed.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`SubscriptionActive`¶

class SubscriptionActive(_TenantScopedLiveAtom)(
    *,
    of: Account = <object object>,
    adapter: Any = None,
) -> None

Subject's subscription is active right now (live, authoritative).

Live counterpart to the claim-based sandloom.shared.authz.atoms_claims.Subscribed. Use this when staleness is unacceptable (e.g. gating immediately after a purchase); use Subscribed for the cheap, zero-I/O common path. of= names the billing account (spec §7).

`SubscriptionActive.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`SubscriptionActive.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

Advanced names¶

`AccessDecision`¶

class AccessDecision(
    allowed: bool,
    code: str = 'ok',
    category: str | None = None,
    reason: str = '',
    http_status: int = 200,
    failed_requirement: str | None = None,
    policy_code: str | None = None,
    metadata: dict[str, Any] = <factory>,
    indeterminate: bool = False,
) -> None

The outcome of evaluating a policy against an AuthzContext.

allowed is the only field that matters for the boolean question; the rest exist so a denial is reportable, auditable and observable.

`AccessDecision.allow`¶

@classmethod
def allow(cls, *, policy_code: str | None = None) -> AccessDecision

`AccessDecision.deny`¶

@classmethod
def deny(
    cls,
    *,
    code: str,
    category: str = 'forbidden',
    reason: str = '',
    http_status: int | None = None,
    failed_requirement: str | None = None,
    policy_code: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> AccessDecision

`AccessDecision.error`¶

@classmethod
def error(
    cls,
    *,
    code: str,
    category: str = 'policy_misconfigured',
    reason: str = '',
    http_status: int | None = None,
    failed_requirement: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> AccessDecision

An indeterminate denial: the requirement could not be evaluated.

Distinct from deny (the subject definitively lacks the requirement): combinators propagate it unchanged, so ~Entitled(x) with x unregistered / its service down DENIES with this category and status instead of inverting the failure into an allow.

`AccessDecision.with_policy_code`¶

def with_policy_code(self, policy_code: str | None) -> AccessDecision

Copy of this decision tagged with policy_code.

`AccessDecision.audit_fields`¶

def audit_fields(self) -> dict[str, Any]

The subset safe and useful to record on a deny (no PII).

`AccessDenied`¶

class AccessDenied(HTTPException)(
    decision: AccessDecision,
    *,
    status_code: int | None = None,
) -> None

Raised by enforce (and the HTTP guards) when a policy is not satisfied.

Carries the full AccessDecision so the caller (HTTP adapter, task runner) can map it to the right transport response. Subclasses FastAPI's HTTPException so a module calling enforce() inside an endpoint gets the decision's status (402/403/501/…) and {"detail", "code"} body from the framework HTTP handler instead of an unhandled 500. Non-HTTP callers (workers, services) still catch the typed AccessDenied and read .decision. It also exposes the framework error-protocol attributes http_status / code / detail.

status_code explicitly overrides the decision's status (transport override, e.g. require(..., status_code=402)).

`AccessFact`¶

class AccessFact(
    name: str,
    owner_module: str,
    claim_key: str | None = None,
    projection_reader: ProjectionReader | None = None,
    live_service: str | None = None,
    live_resolver: LiveResolver | None = None,
    min_freshness: str = 'claim',
    default_mode: str = 'auto',
    deny_category: str | None = None,
    claim_rule: ClaimRule | Callable[[], ClaimRule] | None = None,
) -> None

Declarative binding of a fact name to its three resolution paths.

A module registers one of these in register(core) via core.register_access_fact(...), wiring together mechanisms that already exist: its claims contributor (claim_key), an optional local projection (projection_reader) and its service contract (live_service + live_resolver).

`AccessFact.current_claim_rule`¶

def current_claim_rule(self) -> ClaimRule | None

The declarative claim rule in effect now (None: not declarative).

`AccessFact.interpret_claim`¶

def interpret_claim(self, value: Any) -> bool

Claim value -> bool, exactly as the owner interprets it.

`AccessFact.to_wire`¶

def to_wire(self) -> dict[str, Any]

Wire-safe descriptor published by the owner's describe().

`AccessFact.build_facts_contract`¶

@classmethod
def build_facts_contract(cls, core: Any, owner: str) -> Any

The <owner>.facts.v1 provider (called by the kernel on register).

A duck-typed hook, so the kernel exposes the contract without importing the authorization subsystem.

`Account`¶

class Account(
    kind: AccountKind,
    param: str | None = None,
    reason: str | None = None,
) -> None

The billing account an atom checks (of=); build it with user, org or bound_tenant.

`Account.user`¶

@staticmethod
def user(*, reason: str | None = None) -> Account

The caller's personal account. reason= waives SANDLOOM-C027 on a route that carries an organization parameter.

`Account.org`¶

@staticmethod
def org(*, param: str | None = 'org_id') -> Account

An organization account: param (path, then query), else the X-Tenant-Id header, else the bound tenant (param=None: header or bound tenant only).

`Account.bound_tenant`¶

@staticmethod
def bound_tenant() -> Account

The account of the bound tenant, by its kind (personal or organization; canonical / unknown are indeterminate).

`Account.describe`¶

def describe(self) -> str

`ACTIVE_SUBSCRIPTION_STATUSES`¶

ACTIVE_SUBSCRIPTION_STATUSES: frozenset = frozenset({'active', 'trialing'})

(no docstring)

`And`¶

class And(Requirement)(*parts: Requirement) -> None

Conjunction: all operands must hold. Short-circuits on first non-allow.

A definite deny or an indeterminate operand both stop evaluation and are returned as-is (both deny; the indeterminate keeps its 5xx category).

`And.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`And.decide`¶

async def decide(self, ctx: AuthzContext) -> AccessDecision

Structured result. Default: allow iff evaluate is True.

`And.mentions_resource`¶

def mentions_resource(self) -> bool

True if evaluating this requirement needs the request resource.

Used by require() to reject resource-scoped policies at wiring time (they belong in authorize_resource/authorize), turning a runtime 500 into a clear startup error.

`And.org_params`¶

def org_params(self) -> frozenset[str]

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, X-Tenant-Id, else the bound tenant): create_app then refuses a route that solves the gate before its TenantScope() (SANDLOOM-B007).

`And.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`And.risks`¶

def risks(self) -> frozenset[str]

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): {"admin"} for an admin role, else none. An Or carries its branches' risks; a Not grants no authority.

`And.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`authorize`¶

async def authorize(
    resource: Any,
    policy: Requirement,
    *,
    ctx: UserContext,
    request: Request | None = None,
    db: AsyncSession | None = None,
    status_code: int | None = None,
) -> None

Inline check over an already-loaded resource; raises on failure.

The escape hatch for cases that don't fit authorize_resource (multiple resources, ownership known only after a non-trivial query, authz interleaved with business logic). The endpoint owns the single fetch. Raises AccessDenied with the decision's status (or the explicit status_code override for a definite denial).

`AuthzContext`¶

class AuthzContext(
    subject: UserContext,
    *,
    request: Request | None = None,
    db: AsyncSession | None = None,
    resource: Any = <object object>,
    loader: Callable[[AuthzContext], Awaitable[Any]] | None = None,
    source: str = 'http',
    core: Any = None,
) -> None

Inputs available to a Requirement during evaluation.

`AuthzContext.get_resource`¶

async def get_resource(self) -> Any

Return the request resource, loading it once on first access.

Loading is lazy so that a policy like Perm("x") & Owner(...) never fetches the resource when the cheap claim atom already fails.

`AuthzContext.has_resource`¶

has_resource  (property)

`AuthzContext.core`¶

def core(self) -> Any

Resolve the bound CoreFacade, or None.

Order: an explicitly-passed core (non-HTTP paths pass core= to enforce) → the app the request was served by. There is deliberately no process-global fallback — the subsystem never reaches a global singleton (that would break multi-app / test isolation). When no core is bound, atoms that need one (live / fact / OrgRole) degrade explicitly (indeterminate 501 / FactUnavailable).

`Category`¶

class Category

Mandatory denial categories (TZ §10.2).

`ClaimRule`¶

class ClaimRule(kind: str = 'truthy', values: frozenset[Any] = frozenset()) -> None

Declarative claim -> bool interpretation: wire-safe, no callables.

ClaimRule.one_of({"active", "trialing"}) is True iff the claim value is one of the listed scalars; ClaimRule.truthy() is bool(value). The owner describes its facts with these, so a consumer in another process interprets the token claim EXACTLY like the owner does locally (one source of truth). A rule that depends on the owner's settings (billing's past_due grace) is a zero-argument factory on the fact (AccessFact(claim_rule=factory)), serialized with the owner's current settings at describe() time.

`ClaimRule.one_of`¶

@classmethod
def one_of(cls, values: Iterable[Any]) -> ClaimRule

`ClaimRule.truthy`¶

@classmethod
def truthy(cls) -> ClaimRule

`ClaimRule.interpret`¶

def interpret(self, value: Any) -> bool

`ClaimRule.to_wire`¶

def to_wire(self) -> dict[str, Any]

`ClaimRule.from_wire`¶

@classmethod
def from_wire(cls, raw: Any) -> ClaimRule | None

Parse a wire rule; None for anything not understood.

Fail closed: a consumer that cannot parse the rule never reads the claim and escalates to the owner's live answer instead.

`current_context`¶

async def current_context(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None),
) -> UserContext

Authenticated subject from the Bearer access token (cached per request).

`enforce`¶

async def enforce(
    policy: Requirement,
    *,
    subject: UserContext,
    db: AsyncSession | None = None,
    request: Request | None = None,
    resource: Any = <object object>,
    loader: Callable[[AuthzContext], Awaitable[Any]] | None = None,
    source: str = 'internal',
    core: Any = None,
    policy_code: str | None = None,
) -> AccessDecision

Evaluate policy for subject; raise AccessDenied on deny.

The one chokepoint. HTTP/use-case/task adapters all call this. Returns the allow AccessDecision on success so callers may inspect it.

`Entitled`¶

class Entitled(Requirement)(
    fact_name: str,
    *,
    category: str | None = None,
    code: str | None = None,
    mode: str | None = None,
) -> None

Subject satisfies the registered access fact fact_name.

Args:

fact_name: name of an AccessFact registered via core.register_access_fact(...).
category: denial category when the fact is False. Defaults to the fact's own deny_category (billing's subscription facts declare Category.PAYMENT_REQUIRED → 402), else forbidden (403). A resolver raising FactDenied picks its own category.
mode: force a resolution mode, overriding the ACCESS setting (for tests / special endpoints). Normally left None.

`Entitled.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`Entitled.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Entitled.decide`¶

async def decide(self, ctx: AuthzContext) -> AccessDecision

Structured result. Default: allow iff evaluate is True.

`Entitled.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`ENTITLEMENTS`¶

ENTITLEMENTS: str = 'entitlements'

(no docstring)

`EntitlementsAdapter`¶

class EntitlementsAdapter(Protocol)

Answers subscription / credit questions (SubscriptionActive/Credits).

`EntitlementsAdapter.subscription_active`¶

async def subscription_active(
    self,
    service: Any,
    *,
    tenant_type: str,
    tenant_id: UUID,
    ctx: AuthzContext,
) -> bool

`EntitlementsAdapter.credit_balance`¶

async def credit_balance(
    self,
    service: Any,
    *,
    tenant_type: str,
    tenant_id: UUID,
    ctx: AuthzContext,
) -> int

`FactDenied`¶

class FactDenied(Exception)(
    reason: str,
    *,
    category: str = 'forbidden',
    code: str | None = None,
) -> None

A resolver's definite denial for a reason other than "the fact is False".

Raised by a projection reader / live resolver when the fact is inapplicable to the subject — e.g. organization.subscription.active with no organization in context, or for an org the subject is not a member of. sandloom.shared.authz.atoms_fact.Entitled maps it to a definite deny with category (default forbidden, 403) instead of the fact's deny_category — "you may not ask" is not "pay to unlock".

`FactOwnerUnreachable`¶

class FactOwnerUnreachable(FactUnavailable)

The fact's owner is bound but could not describe its facts (503).

Distinct from "not configured" (501): the deployment is correct, the owner is down. Never a silent allow or a definite deny.

`FactUnavailable`¶

class FactUnavailable(Exception)

A fact could not be resolved in any acceptable mode (a misconfiguration).

Raised when an explicitly-requested mode cannot be honored (live service not bound, no projection reader) or auto exhausts every option. The atom maps this to a policy_misconfigured decision — never a silent allow.

`Feature`¶

class Feature(_AccountClaimAtom)(
    key: str,
    *,
    of: Account = <object object>,
    claim_key: str = 'features',
) -> None

The account's plan enables the feature flag key (of=, §7.2).

`Feature.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`lookup_access_fact`¶

async def lookup_access_fact(ctx: AuthzContext, name: str) -> AccessFact | None

The fact name: the local registration, else the owner's public one.

Order: core.get_access_fact(name) (owner co-deployed) → every bound <owner>.facts.v1 with no local provider (the owner named by the fact's first dotted segment first), whose describe() is cached on the core for REMOTE_FACT_TTL_SECONDS. None when no owner knows the fact (the caller reports 501); FactOwnerUnreachable (503) when a bound owner could not be asked and none of the others knows it.

`make_rebuild_gate`¶

def make_rebuild_gate(rebuilder: Any, *, name: str = '') -> StartupTask

A startup task that rebuilds an authz projection from its sources.

Captures the rebuilder supplied at registration in its closure (no process-global / registry lookup at run time). Install as critical=True so the process refuses traffic until the projection is populated — otherwise an empty table makes every fact resolve False and the consumer fail-closed-denies all callers. The rebuilder must be idempotent (the protocol guarantees it), so re-running on boot is safe; for large projections prefer sandloom projection rebuild <name> in the one-shot migrate step with rebuild_on_startup=False.

`MEMBERSHIP`¶

MEMBERSHIP: str = 'membership'

(no docstring)

`MembershipAdapter`¶

class MembershipAdapter(Protocol)

Answers organization-role questions for OrgRole.

`MembershipAdapter.has_role_at_least`¶

async def has_role_at_least(
    self,
    service: Any,
    *,
    org_id: UUID,
    user_id: UUID,
    minimum: str,
    ctx: AuthzContext,
) -> bool

`MODE_AUTO`¶

MODE_AUTO: str = 'auto'

(no docstring)

`MODE_CLAIM`¶

MODE_CLAIM: str = 'claim'

(no docstring)

`MODE_LIVE`¶

MODE_LIVE: str = 'live'

(no docstring)

`MODE_PROJECTION`¶

MODE_PROJECTION: str = 'projection'

(no docstring)

`Not`¶

class Not(Requirement)(inner: Requirement) -> None

Negation of a single requirement.

Negates only a definite answer: an indeterminate inner decision is returned unchanged (deny), so ~Entitled("account.suspended") never allows because the fact is unregistered or its service is down.

`Not.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Not.decide`¶

async def decide(self, ctx: AuthzContext) -> AccessDecision

Structured result. Default: allow iff evaluate is True.

`Not.mentions_resource`¶

def mentions_resource(self) -> bool

True if evaluating this requirement needs the request resource.

Used by require() to reject resource-scoped policies at wiring time (they belong in authorize_resource/authorize), turning a runtime 500 into a clear startup error.

`Not.org_params`¶

def org_params(self) -> frozenset[str]

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, X-Tenant-Id, else the bound tenant): create_app then refuses a route that solves the gate before its TenantScope() (SANDLOOM-B007).

`Not.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`Not.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`Or`¶

class Or(Requirement)(*parts: Requirement) -> None

Disjunction: any operand may hold. Short-circuits on first allow.

Only a genuinely satisfied operand allows — and one that does allows even if a sibling is indeterminate (Role("admin") | Entitled("x") admits an admin while the x fact is unregistered): the grant comes from the satisfied operand, never from inverting or guessing the broken one. When none allows and at least one operand was indeterminate, the disjunction is indeterminate (the first such decision is returned, preserving its category/status) — a broken branch is reported, not silently folded into a plain 403.

`Or.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Or.decide`¶

async def decide(self, ctx: AuthzContext) -> AccessDecision

Structured result. Default: allow iff evaluate is True.

`Or.mentions_resource`¶

def mentions_resource(self) -> bool

True if evaluating this requirement needs the request resource.

Used by require() to reject resource-scoped policies at wiring time (they belong in authorize_resource/authorize), turning a runtime 500 into a clear startup error.

`Or.org_params`¶

def org_params(self) -> frozenset[str]

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, X-Tenant-Id, else the bound tenant): create_app then refuses a route that solves the gate before its TenantScope() (SANDLOOM-B007).

`Or.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`Or.risks`¶

def risks(self) -> frozenset[str]

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): {"admin"} for an admin role, else none. An Or carries its branches' risks; a Not grants no authority.

`Or.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`Perm`¶

class Perm(Requirement)(perm: str) -> None

Subject holds the fine-grained permission perm.

The permission is stored as perm — code stays the machine-readable denial code of the atom ("Perm"), not the permission name.

`Perm.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Perm.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`Permissions`¶

class Permissions

Predefined fine-grained permission codes.

`PolicyIndeterminate`¶

class PolicyIndeterminate(Exception)(decision: AccessDecision) -> None

Raised by an atom's evaluate when it cannot answer at all.

evaluate is boolean, so an atom that cannot evaluate (service absent, fact unregistered, no tenant context) must not answer False — a caller composing not await atom.evaluate(ctx) would turn the failure into an allow. It raises this instead; Requirement.decide (and enforce) convert it into the carried indeterminate AccessDecision. Never an HTTPException — atoms run in workers too.

`register_authz_projection`¶

def register_authz_projection(
    core: Any,
    *,
    descriptor: Any,
    consumers: list[Any],
    rebuilder: Any,
    rebuild_on_startup: bool = False,
    critical: bool = True,
) -> None

Register an authorization projection, its consumers and rebuilder.

descriptor is a QueryProjectionDescriptor; consumers a list of ProjectionConsumerDescriptor (one per source channel); rebuilder a ProjectionRebuilder. When rebuild_on_startup is True a rebuild gate is installed (critical controls whether its failure aborts startup).

`register_live_adapter`¶

def register_live_adapter(kind: str, adapter: Any) -> None

Bind the adapter the live atoms use for kind — CONSUMER-side config.

kind is ENTITLEMENTS or MEMBERSHIP; adapter must satisfy EntitlementsAdapter / MembershipAdapter.

The table is per PROCESS and is read where the atom is EVALUATED — the consumer. Register an adapter from the consumer's composition root or the consuming module's register(core), in every process that evaluates Credits / SubscriptionActive / OrgRole. A provider module that registers an adapter for its OWN contract only affects processes that load it: in a split deployment the consumer process silently keeps the default — so that is logged as a warning.

`Requirement`¶

class Requirement(ABC)

An authorization predicate, composable with & | ~.

evaluate is the primitive boolean check (it raises PolicyIndeterminate when it cannot answer). decide wraps it into a structured AccessDecision (the engine/enforcement path uses decide so a denial carries a machine-readable code/category and the failing requirement); the two never disagree.

`Requirement.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Requirement.decide`¶

async def decide(self, ctx: AuthzContext) -> AccessDecision

Structured result. Default: allow iff evaluate is True.

`Requirement.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`Requirement.mentions_resource`¶

def mentions_resource(self) -> bool

True if evaluating this requirement needs the request resource.

Used by require() to reject resource-scoped policies at wiring time (they belong in authorize_resource/authorize), turning a runtime 500 into a clear startup error.

`Requirement.org_params`¶

def org_params(self) -> frozenset[str]

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, X-Tenant-Id, else the bound tenant): create_app then refuses a route that solves the gate before its TenantScope() (SANDLOOM-B007).

`Requirement.contracts`¶

def contracts(self) -> frozenset[str]

Sync contracts evaluating this requirement may call (wiring time).

A module's derived consumes includes them (optional), so the startup preflight reports an unbound one and sandloom add deploy binds it in a split deployment. Default: none (a claims-only atom).

`Requirement.risks`¶

def risks(self) -> frozenset[str]

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): {"admin"} for an admin role, else none. An Or carries its branches' risks; a Not grants no authority.

`resolve_fact`¶

async def resolve_fact(
    fact: AccessFact,
    ctx: AuthzContext,
    *,
    mode: str | None = None,
) -> bool

Resolve fact to a boolean, caching the result on ctx.

mode overrides the configured mode (used by tests and to force a path); otherwise the mode comes from settings, clamped to the fact's floor.

`Role`¶

class Role(Requirement)(role: str) -> None

Subject holds the global role role.

The role is stored as role — code stays the machine-readable denial code of the atom ("Role"), not the role name.

`Role.evaluate`¶

async def evaluate(self, ctx: AuthzContext) -> bool

Return True iff the subject satisfies this requirement.

Raise PolicyIndeterminate (never return False) when the requirement cannot be evaluated at all.

`Role.risks`¶

def risks(self) -> frozenset[str]

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): {"admin"} for an admin role, else none. An Or carries its branches' risks; a Not grants no authority.

`Role.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

`Roles`¶

class Roles

Predefined global role codes.

`service_subject`¶

def service_subject(*, roles: frozenset[str] = frozenset({'service'})) -> UserContext

A subject representing another service calling across a module boundary.

`SOURCE_HTTP`¶

SOURCE_HTTP: str = 'http'

(no docstring)

`SOURCE_INTERNAL`¶

SOURCE_INTERNAL: str = 'internal'

(no docstring)

`SOURCE_SYSTEM`¶

SOURCE_SYSTEM: str = 'system'

(no docstring)

`SOURCE_TASK`¶

SOURCE_TASK: str = 'task'

(no docstring)

`system_subject`¶

def system_subject(*, roles: frozenset[str] = frozenset({'system'})) -> UserContext

A subject representing the platform itself (trusted background work).

`TIER_CLAIM`¶

TIER_CLAIM: int = 1

(no docstring)

`TIER_LIVE`¶

TIER_LIVE: int = 3

(no docstring)

`TIER_RESOURCE`¶

TIER_RESOURCE: int = 2

(no docstring)

`WithinLimit`¶

class WithinLimit(_AccountClaimAtom)(
    key: str,
    used: int,
    *,
    of: Account = <object object>,
    claim_key: str = 'limits',
) -> None

used is strictly below the account's plan limit for key.

For Account.user() the limit comes from the token claim; used is the current count the caller already knows (e.g. number of seats). Absent / non-integer limit ⇒ treated as 0 (i.e. denied), so a missing entitlement fails closed.

`WithinLimit.describe`¶

def describe(self) -> str

Human-readable label used in 403 detail messages.

sandloom.sdk.authz¶

Start with (beginner)¶

Advanced¶

Beginner names¶

authorize_resource¶

Credits¶

Credits.evaluate¶

Credits.describe¶

OrgRole¶

OrgRole.org_params¶

OrgRole.risks¶

OrgRole.contracts¶

OrgRole.adapter¶

OrgRole.evaluate¶

OrgRole.describe¶

Owner¶

Owner.evaluate¶

Owner.describe¶

require¶

Subscribed¶

Subscribed.tenant_type¶

Subscribed.org_params¶

Subscribed.contracts¶

Subscribed.evaluate¶

Subscribed.describe¶

SubscriptionActive¶

SubscriptionActive.evaluate¶

SubscriptionActive.describe¶

Advanced names¶

AccessDecision¶

AccessDecision.allow¶

AccessDecision.deny¶

AccessDecision.error¶

AccessDecision.with_policy_code¶

AccessDecision.audit_fields¶

AccessDenied¶

AccessFact¶

AccessFact.current_claim_rule¶

AccessFact.interpret_claim¶

AccessFact.to_wire¶

AccessFact.build_facts_contract¶

Account¶

Account.user¶

Account.org¶

Account.bound_tenant¶

Account.describe¶

ACTIVE_SUBSCRIPTION_STATUSES¶

And¶

And.evaluate¶

And.decide¶

And.mentions_resource¶

And.org_params¶

And.contracts¶

And.risks¶

And.describe¶

authorize¶

AuthzContext¶

AuthzContext.get_resource¶

AuthzContext.has_resource¶

AuthzContext.core¶

Category¶

ClaimRule¶

ClaimRule.one_of¶

ClaimRule.truthy¶

ClaimRule.interpret¶

ClaimRule.to_wire¶

ClaimRule.from_wire¶

current_context¶

enforce¶

Entitled¶

Entitled.contracts¶

Entitled.evaluate¶

Entitled.decide¶

Entitled.describe¶

ENTITLEMENTS¶

EntitlementsAdapter¶

EntitlementsAdapter.subscription_active¶

EntitlementsAdapter.credit_balance¶

FactDenied¶

FactOwnerUnreachable¶

`sandloom.sdk.authz`¶

`authorize_resource`¶

`Credits`¶

`Credits.evaluate`¶

`Credits.describe`¶

`OrgRole`¶

`OrgRole.org_params`¶

`OrgRole.risks`¶

`OrgRole.contracts`¶

`OrgRole.adapter`¶

`OrgRole.evaluate`¶

`OrgRole.describe`¶

`Owner`¶

`Owner.evaluate`¶

`Owner.describe`¶

`require`¶

`Subscribed`¶

`Subscribed.tenant_type`¶

`Subscribed.org_params`¶

`Subscribed.contracts`¶

`Subscribed.evaluate`¶

`Subscribed.describe`¶

`SubscriptionActive`¶

`SubscriptionActive.evaluate`¶

`SubscriptionActive.describe`¶

`AccessDecision`¶

`AccessDecision.allow`¶

`AccessDecision.deny`¶

`AccessDecision.error`¶

`AccessDecision.with_policy_code`¶

`AccessDecision.audit_fields`¶

`AccessDenied`¶

`AccessFact`¶

`AccessFact.current_claim_rule`¶

`AccessFact.interpret_claim`¶

`AccessFact.to_wire`¶

`AccessFact.build_facts_contract`¶

`Account`¶

`Account.user`¶

`Account.org`¶

`Account.bound_tenant`¶

`Account.describe`¶

`ACTIVE_SUBSCRIPTION_STATUSES`¶

`And`¶

`And.evaluate`¶

`And.decide`¶

`And.mentions_resource`¶

`And.org_params`¶

`And.contracts`¶

`And.risks`¶

`And.describe`¶

`authorize`¶

`AuthzContext`¶

`AuthzContext.get_resource`¶

`AuthzContext.has_resource`¶

`AuthzContext.core`¶

`Category`¶

`ClaimRule`¶

`ClaimRule.one_of`¶

`ClaimRule.truthy`¶

`ClaimRule.interpret`¶

`ClaimRule.to_wire`¶

`ClaimRule.from_wire`¶

`current_context`¶

`enforce`¶

`Entitled`¶

`Entitled.contracts`¶

`Entitled.evaluate`¶

`Entitled.decide`¶

`Entitled.describe`¶

`ENTITLEMENTS`¶

`EntitlementsAdapter`¶

`EntitlementsAdapter.subscription_active`¶

`EntitlementsAdapter.credit_balance`¶

`FactDenied`¶

`FactOwnerUnreachable`¶

`FactUnavailable`¶

`Feature`¶

`Feature.describe`¶

`lookup_access_fact`¶