Skip to content

API Reference

The public surface, generated from the source docstrings.

Core

purview.Policy

A registry of authorization rules plus tenant-scoping configuration.

Rules are keyed by (model, action). Multiple rules may be registered for the same key; their predicates are OR-combined (rules are granting).

global_models property 

global_models: frozenset[type]

The set of models opted out of tenant scoping.

rule 

rule(model: type, action: str) -> Callable[[RuleFn], RuleFn]

Register a predicate function for (model, action).

Used as a decorator::

@policy.rule(Post, "read")
def read_post(ctx: Context) -> list[ColumnElement[bool]]:
    rules = []
    if ctx.has_role("author"):
        rules.append(Post.author_id == ctx.user_id)
    return rules  # OR-combined; empty list = deny

rules_for 

rules_for(model: type, action: str) -> list[RuleFn]

All registered rule functions for (model, action) (possibly empty).

has_rules 

has_rules(model: type, action: str) -> bool

Whether any rule is registered for (model, action).

create_rule 

create_rule(model: type) -> Callable[[CreateRuleFn], CreateRuleFn]

Register a create rule for model.

create has no row to filter, so its rule is a plain predicate over the proposed instance — type-checked and unit-testable like any function::

@policy.create_rule(Post)
def create_post(ctx: Context, post: Post) -> bool:
    return post.author_id == ctx.user_id

Multiple create rules for one model are AND-combined (all must pass). The tenant is validated structurally regardless, and the write guard remains the backstop at flush.

create_rules_for 

create_rules_for(model: type) -> list[CreateRuleFn]

All registered create rules for model (possibly empty).

global_model 

global_model(model: type) -> type

Mark model as global (NOT tenant-scoped).

Secure-by-default: every model is tenant-scoped unless explicitly marked global here. Usable as a decorator, returning the class unchanged.

is_global 

is_global(model: type) -> bool

Whether model has been opted out of tenant scoping.

set_tenant_field 

set_tenant_field(model: type, column: str) -> None

Override the tenant column name for model (default is per-install).

tenant_field_for 

tenant_field_for(model: type, default: str) -> str

The tenant column for model, falling back to the install default.

Walks the MRO, so a subclass inherits an override registered on its base.

role_implies 

role_implies(role: str, *implied: str) -> None

Declare that holding role also grants implied roles.

Implications are transitive: role_implies("admin", "editor") and role_implies("editor", "viewer") together mean an admin holds editor and viewer too. The closure is applied once when a context is bound, so rules keep calling ctx.has_role("viewer") unchanged and a higher role satisfies it automatically.

expand_roles 

expand_roles(roles: Iterable[str]) -> frozenset[str]

The transitive closure of roles under the declared implications.

Cycle- and self-loop-safe. With no implications declared (or none reachable from roles), returns the input as a frozenset unchanged.

purview.Context dataclass

Bases: Generic[U, T]

An authenticated actor, scoped to one tenant.

Parameters

user_id: Identifier of the acting principal. Flows into predicates as a bind value (e.g. Post.author_id == ctx.user_id). tenant_id: The single tenant this context is scoped to. roles: The actor's roles within this tenant. Any iterable of strings is accepted and normalised to a frozenset.

has_role 

has_role(name: str) -> bool

True if the actor holds name in this tenant.

has_any 

has_any(*names: str) -> bool

True if the actor holds at least one of names.

Enforcement (purview.sqlalchemy)

purview.sqlalchemy.install 

install(base: Any, policy: Policy, *, tenant_column: str = 'tenant_id', strict: bool = False, warn_on_unfiltered: bool = False, audit: _AuditMode = 'off', session_class: type[Session] = Session) -> Purview

Discover scoped models, wire the guards, and return the enforcer.

Set strict=True to deny reads of a scoped model that has no read rule (within-tenant default deny) instead of defaulting it to tenant-scope-only.

Set warn_on_unfiltered=True to emit :class:~purview.exceptions.PurviewWarning for the documented sharp edges: queries on an unbound session and raw/non-ORM statements on a bound one. Off by default (no behavior change).

Set audit to "warn" or "raise" to surface scoped models with no read rule (visible tenant-wide) at install time; "raise" raises :class:~purview.exceptions.PolicyAuditError. Under strict=True such models deny by default, so the audit finds nothing.

Raises :class:~purview.exceptions.UnscopedModel if a non-global model lacks the tenant column.

purview.sqlalchemy.Purview

Holds a policy + tenant configuration and enforces it on a session class.

Construct via :func:install (which also wires the guards). The same object answers object checks, batch checks, and create validation, all derived from the one policy.

install 

install() -> Purview

Wire the read/write/attach guards onto the session class (idempotent).

uninstall 

uninstall() -> None

Remove the guards from the session class.

bind 

bind(session: _SessionLike, ctx: Context[Any, Any]) -> None

Bind ctx to session so reads filter and writes are guarded.

Declared role implications (:meth:Policy.role_implies) are expanded once here, so every rule, check, and explain sees the actor's effective roles.

context 

context(session: _SessionLike) -> Context[Any, Any]

The context bound to session (raises if the session is unbound).

explain 

explain(session_or_ctx: _SessionLike | Context[Any, Any], action: str, model: type) -> PredicateExplanation

Explain the tenant + row predicate for (model, action).

Accepts a bound session or a bare :class:~purview.core.context.Context. Pure introspection — it compiles the predicate the guard would apply and never touches the database. Role implications are expanded first.

audit 

audit() -> AuditReport

Classify the read visibility of every model (pure, no database access).

Reports scoped models with no read rule — visible tenant-wide under the default policy. See :meth:AuditReport.tenant_wide_models.

authorize async 

authorize(session: AsyncSession, action: str, resource: object) -> bool

Whether the bound actor may perform action on resource.

authorized_ids async 

authorized_ids(session: AsyncSession, action: str, model: type, ids: Iterable[Any]) -> list[Any]

The subset of ids the bound actor may perform action on.

validate_create 

validate_create(session: _SessionLike, resource: object) -> bool

Whether resource may be created in the bound actor's tenant.

Checks the proposed tenant and every registered create_rule for the model. The write guard remains the structural backstop at flush.

purview.sqlalchemy.authorized_select 

authorized_select(policy: Policy, ctx: Context[Any, Any], model: type, tenant_column: str, strict: bool = False) -> Select[Any]

A select(model) narrowed to the rows ctx may read.

Applies tenant scope and the read predicate explicitly. On a bound session the guard would apply equivalent criteria too; the duplication is harmless.

purview.sqlalchemy.bypass

The escape hatch.

A single, loud, greppable bypass for admin tooling and migrations. While active on the current task, both the read guard and the write guard stand down.

bypass is implemented with a :class:contextvars.ContextVar, so it scopes to the current (async) task and never bleeds across requests.

is_bypassed 

is_bypassed() -> bool

Whether enforcement is currently suppressed on this task.

bypass 

bypass(reason: str) -> Iterator[None]

Suspend all Purview enforcement within the block.

A non-empty reason is required and logged at WARNING — bypasses are meant to be visible in logs and greppable in code::

with policy.bypass(reason="nightly billing rollup"):
    ...

Predicate helpers (purview.predicates)

purview.predicates.owned_by 

owned_by(column: InstrumentedAttribute[Any], ctx: Context[Any, Any]) -> ColumnElement[bool]

column == ctx.user_id — the canonical "I own this row" predicate.

Note: an anonymous context (user_id is None) yields column IS NULL; deny anonymous actors upstream rather than relying on this.

purview.predicates.in_values 

in_values(column: InstrumentedAttribute[Any], values: Iterable[Any]) -> ColumnElement[bool]

column IN (values) — membership against a precomputed set.

Use for sets resolved when the context was built (e.g. the team ids an actor belongs to). An empty values compiles to false() — preserving default-deny and avoiding empty-IN dialect warnings.

Introspection

purview.PredicateExplanation dataclass

What the guard would apply for one (model, action) under one context.

purview.RuleContribution dataclass

One registered rule's contribution to the combined row predicate.

purview.AuditReport dataclass

The visibility classification of every discovered model.

tenant_wide_models property 

tenant_wide_models: tuple[ModelAudit, ...]

Scoped models readable by every actor in their tenant (no read rule).

purview.ModelAudit dataclass

The read-visibility classification of one model.

FastAPI (purview.fastapi)

purview.fastapi.context_binder 

context_binder(pv: Purview, session_dependency: Callable[..., Any], context_dependency: Callable[..., Any]) -> Callable[..., Awaitable[AsyncSession]]

A dependency that binds the resolved context to the request session.

Inject the result into routes to obtain a session whose reads are already scoped to the actor::

bound = context_binder(pv, get_session, get_context)

@app.get("/posts")
async def list_posts(session: AsyncSession = Depends(bound)):
    return (await session.scalars(select(Post))).all()

purview.fastapi.requires 

requires(pv: Purview, action: str, resource_type: type, context_dependency: Callable[..., Any]) -> Callable[..., Awaitable[None]]

A route dependency that 403s an actor with no standing grant for the action.

Collection routes still filter per-row afterwards; this only short-circuits actors who categorically cannot perform action on resource_type.

purview.fastapi.authorize_or_403 async 

authorize_or_403(pv: Purview, session: AsyncSession, action: str, resource: R) -> R

Return resource if the bound actor may action it, else 403.

purview.fastapi.install_error_handlers 

install_error_handlers(app: FastAPI) -> None

Register 403 handlers for :class:PurviewForbidden and :class:CrossTenantWrite.

Both represent an actor attempting something they are not permitted to do — a denied read/action, or a write into another tenant.

Exceptions

purview.PurviewError

Bases: Exception

Base class for every error raised by Purview.

purview.PurviewForbidden

Bases: PurviewError

An actor is not permitted to perform an action on a resource.

The FastAPI adapter translates this into an HTTP 403 response.

purview.CrossTenantWrite

Bases: PurviewError

A flush would write a row into a tenant other than the session's.

Raised by the write guard for forged-tenant inserts and for updates that move an existing row across the tenant boundary.

purview.TenantMismatch

Bases: PurviewError

A session bound to one tenant is being rebound to a different one.

Raised by bind_context / Purview.bind when a request would reuse a session that already carries another tenant's context — the cross-tenant footgun the session boundary exists to prevent.

purview.UnscopedModel

Bases: PurviewError

A non-global model lacks a resolvable tenant column.

Raised at install() time (fail closed) so a model that would otherwise ship unscoped is rejected at startup rather than leaking at query time.

purview.PolicyAuditError

Bases: PurviewError

The policy audit found scoped models that are visible tenant-wide.

Raised at install(audit="raise") time so a model with no read rule — readable by every actor in its tenant — is surfaced at startup rather than discovered in production.

purview.PurviewWarning

Bases: UserWarning

An opt-in developer warning about a likely enforcement gap.

Never emitted by default. install(warn_on_unfiltered=True) turns on the runtime warnings (raw text() on a bound session, queries on an unbound session) and install(audit="warn") reports tenant-wide models. These are advisory aids, not controls — the enforcement boundary is unchanged.