Row-level authorization and multi-tenancy for FastAPI + SQLAlchemy. Define a policy once as SQLAlchemy column expressions and get both yes/no checks and query filtering from the same rule — so the check and the filter can never disagree.
Drift between "can this actor do this?" and "which rows can they see?" is a data leak. Purview makes both come from one definition, so they cannot drift.
@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) # authors see their own
if ctx.has_role("org_admin"):
rules.append(true()) # admins see the whole tenant
return rules # OR-combined; empty = denyThat one rule now powers a filtered select(Post) and an
authorize(session, "read", post) check.
Authentication generalizes; authorization does not, because it is welded to your domain model and data layer. The hard, valuable part is data filtering: shaping queries so a user only ever loads rows they're allowed to see, pushed into SQL rather than filtered in Python after the fact. Oso solved this well and then deprecated its open-source library, leaving no idiomatic Python answer. Purview targets that gap for the FastAPI + SQLAlchemy stack specifically — in-process, async-first, no external policy service.
pip install purview-authz # core + SQLAlchemy
pip install "purview-authz[fastapi]" # plus the FastAPI adapterThe distribution is purview-authz; the import package is purview. Requires
Python 3.11+ and SQLAlchemy 2.0+.
from sqlalchemy import true
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from purview import Context, Policy, READ
from purview.sqlalchemy import install
class Base(DeclarativeBase): ...
class Org(Base): # the tenant root — global
__tablename__ = "org"
id: Mapped[int] = mapped_column(primary_key=True)
class Post(Base): # tenant-scoped (has the tenant column)
__tablename__ = "post"
id: Mapped[int] = mapped_column(primary_key=True)
org_id: Mapped[int] = mapped_column()
author_id: Mapped[int] = mapped_column()
policy = Policy()
policy.global_model(Org) # opt Org out of tenant scoping
@policy.rule(Post, READ)
def read_post(ctx: Context):
return [Post.author_id == ctx.user_id] if ctx.has_role("author") else []
pv = install(Base, policy, tenant_column="org_id") # wires the guards; validates modelsBind a request's session to its actor, then query normally — reads are filtered automatically:
async with async_session() as session:
pv.bind(session, Context(user_id=42, tenant_id=1, roles={"author"}))
posts = await session.scalars(select(Post)) # only org 1 + authored by 42
one = await session.get(Post, 99) # None if not visible
ok = await pv.authorize(session, "update", post) # yes/no for one object
ids = await pv.authorized_ids(session, "read", Post, [1, 2, 3]) # the allowed subsetOne definition, two forms. A rule returns boolean ColumnElement predicates.
As a .where(...) they filter a collection; wrapped in
EXISTS (SELECT 1 ... AND <predicate>) they check a single object. The database
evaluates both, so relationship and join predicates work without re-implementing
SQL in Python.
Roles select predicates. An actor's tenant-scoped roles decide which
predicates apply for (action, model). Grants are OR-combined; no granting role
means no rows — default deny.
Tenancy is the session boundary. One session is bound to exactly one tenant.
A do_orm_execute hook scopes every read (including lazy and eager relationship
loads) to that tenant; a before_flush hook auto-stamps the tenant on inserts and
refuses writes that would cross the boundary.
Secure by default. Every mapped model is tenant-scoped automatically. Opt a
model out with policy.global_model(...). install() raises if a non-global
model lacks the tenant column — an unscoped table fails at startup, never leaks at
runtime.
read drives filtering. It is the action that shapes collections. Other
actions are instance-level checks; update/delete reuse the read predicate
unless you register a stricter rule for them.
Within-tenant default. A scoped model with no read rule is visible
tenant-wide (tenant isolation still applies). Pass install(..., strict=True) to
flip this to within-tenant default deny — every model then needs an explicit
rule to grant any access. The cross-tenant boundary is enforced identically in
both modes.
Predicate helpers for the common patterns:
from purview.predicates import owned_by, in_values
@policy.rule(Post, READ)
def read_post(ctx):
rules = [owned_by(Post.author_id, ctx)] # Post.author_id == ctx.user_id
if ctx.has_role("editor"):
rules.append(in_values(Post.section_id, ctx.sections)) # empty -> deny
return rulesRole hierarchies — a higher role grants the lower ones, expanded once at bind:
policy.role_implies("admin", "editor") # admin satisfies has_role("editor")Explain the exact predicate the guard would apply — no database round-trip:
print(pv.explain(session, "read", Post)) # or pass a bare Context
# tenant scope : post.org_id = 1
# row predicate : post.author_id = 42Audit for models left visible tenant-wide, and warn on the unfiltered sharp edges — both opt-in:
pv = install(Base, policy, tenant_column="org_id",
audit="warn", # report scoped models with no read rule
warn_on_unfiltered=True) # PurviewWarning on raw text()/unbound queries
pv.audit().tenant_wide_models # inspect the same report at any timeInside the boundary: ORM selects, session.get, relationship loads, and flushes
on a bound session.
Outside the boundary (documented, not enforced):
- Raw SQL and Core
text()— Purview shapes ORM statements, not hand-written SQL. - Implicit lazy loads under async — these raise
MissingGreenletin SQLAlchemy regardless; useselectinload(...)orawait obj.awaitable_attrs.x. Eager and awaitable lazy loads are filtered. - Unbound sessions — a session with no bound context is not filtered (this is how you seed and run migrations).
One loud, greppable bypass for admin tooling and migrations:
from purview.sqlalchemy import bypass
with bypass(reason="nightly billing rollup"):
... # enforcement stands down on this task; the reason is logged at WARNINGfrom purview.fastapi import context_binder, authorize_or_403, install_error_handlers
install_error_handlers(app) # PurviewForbidden -> 403
bound = context_binder(pv, get_session, get_context) # binds the actor per request
@app.get("/posts")
async def list_posts(session: AsyncSession = Depends(bound)):
return (await session.scalars(select(Post))).all() # auto-filtered
@app.patch("/posts/{post_id}")
async def edit(post_id: int, session: AsyncSession = Depends(bound)):
post = await session.get(Post, post_id) # 404 if not visible
await authorize_or_403(pv, session, "update", post) # 403 if not permitted
...See tests/examples/test_blog_app.py for a
complete, runnable app.
Full docs: https://jestatsio.github.io/purview/
- Design — the architecture and the keystone idea.
- Threat model — what is and isn't enforced, each guarantee mapped to the test that proves it.
- Migrating from Oso.
- Example app — a multi-tenant tracker (FastAPI + Alembic + Postgres) exercising per-model columns, composite/UUID keys, and read/create rules.
- Security policy · Contributing · Releasing.
| Purview | Oso (OSS) | Cerbos | Casbin | |
|---|---|---|---|---|
| In-process (no network) | ✅ | ✅ | ❌ service | ✅ |
| SQL data filtering | ✅ | ✅ | ✅ | ❌ |
| One def → check + filter | ✅ | ✅ | ➖ | ❌ |
| SQLAlchemy 2.0 async | ✅ | ❌ | ✅ adapter | ➖ |
| Policy in native Python | ✅ | Polar DSL | YAML | model+CSV |
| Maintained | ✅ | deprecated 2023 | ✅ | ✅ |
In: row-level filtering, multi-tenancy as a structural concern, yes/no checks and query filtering from one definition, SQLAlchemy 2.0 async, FastAPI adapter.
Out (for now): field-level authorization (belongs in serialization), non-SQLAlchemy ORMs, a hosted policy service, Postgres RLS as a compile target.
uv run --extra dev pytest # unit + integration + the example app
uv run --extra dev mypy # strict typing is a project invariant
uv run --extra dev ruff check .Postgres fidelity is exercised in CI; set PURVIEW_TEST_POSTGRES_URL to run the
integration matrix against a local Postgres too.
Releases publish to PyPI on a version tag via Trusted Publishing (no stored token); the version is derived from the tag. See RELEASING.md.
MIT — see LICENSE.