Quickstart

Run a complete example with an in-memory SQLite database. It creates three posts, binds an author to tenant 1, and shows that only their own post is returned.

1. Install

uv init purview-demo
cd purview-demo
uv add "purview-authz[sqlite]"

Prefer pip? Follow the virtual environment instructions and use python quickstart.py below.

2. Add the example

Save this as quickstart.py, or copy it from the repository. This page includes that file directly, so the example and documentation stay in sync.

quickstart.py

"""Run with Python after installing purview-authz[sqlite]."""

import asyncio

from sqlalchemy import ColumnElement, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from purview import READ, Context, Policy
from purview.sqlalchemy import install


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "post"
    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int]
    author_id: Mapped[int]
    title: Mapped[str]


policy = Policy()


@policy.rule(Post, READ)
def read_posts(ctx: Context[int, int]) -> list[ColumnElement[bool]]:
    return [Post.author_id == ctx.user_id] if ctx.has_role("author") else []


async def main() -> None:
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    pv = install(Base, policy, strict=True)
    try:
        async with engine.begin() as connection:
            await connection.run_sync(Base.metadata.create_all)

        # A separate, unbound session for trusted setup.
        async with sessions() as session:
            session.add_all(
                [
                    Post(id=1, tenant_id=1, author_id=42, title="My draft"),
                    Post(id=2, tenant_id=1, author_id=7, title="Another author"),
                    Post(id=3, tenant_id=2, author_id=42, title="Another tenant"),
                ]
            )
            await session.commit()

        async with sessions() as session:
            pv.bind(session, Context(user_id=42, tenant_id=1, roles={"author"}))
            posts = (await session.scalars(select(Post))).all()
            print([post.title for post in posts])  # ['My draft']
            print(await session.get(Post, 3))  # None
            print(await pv.authorize(session, READ, posts[0]))  # True
    finally:
        pv.uninstall()
        await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

3. Run it

uv run python quickstart.py

Expected output:

['My draft']
None
True

The other author's post fails the read policy. The third post belongs to another tenant, even though its author ID matches. Both are filtered out.

What happened

1. Define a model. Post.tenant_id is the tenant field. All mapped models under Base must have this field unless explicitly marked global.
2. Register a rule. Authors contribute an ownership predicate. Other actors contribute no predicates, so the rule denies access.
3. Install once. install() validates the mapped models and registers SQLAlchemy session event handlers. Call it after importing your application's models.
4. Bind before querying. A fresh session receives the authenticated actor's Context. The read guard adds both tenant and row predicates.
5. Check an object explicitly. authorize() uses a database EXISTS query for the object's primary key, tenant, and action predicate.

The separate seed session is intentionally unbound. Unbound sessions have no automatic filtering, which is useful for trusted setup work. Request handlers should always use a fresh bound session.

Choose your default

The example uses strict=True. This denies reads for scoped models that have no read rule. In the default strict=False mode, a scoped model with no read rule is visible to all actors within the bound tenant. In both modes, a registered rule that returns no granting predicates denies access.

Add write permissions

authorize(session, "update", post) checks update permission. When there is no update rule, Purview falls back to the read rule. Define an update rule when editing should be more restrictive than reading, then check before changing the object:

from purview import PurviewForbidden

if not await pv.authorize(session, "update", post):
    raise PurviewForbidden("You cannot edit this post")
post.title = "Revised draft"
await session.commit()

The tenant flush guard does not automatically call your update, delete, or create rules. Use explicit helpers for those permissions. See writing policies for create validation and FastAPI integration for complete request-handler patterns.