SQLAlchemy Local Authorization Integration

Use SQLAlchemy to execute local authorization queries.

Our open-source integration for SQLAlchemy provides ORM-native support for Local Authorization.

pip install sqlalchemy-oso-cloud

Modeling

The SQLAlchemy integration automatically figures out your Local Authorization configuration from annotations on your models, so you can specify your facts entirely within the ORM.

Use the sqlalchemy_oso_cloud.orm.Resource mixin to specify that a model corresponds to an authorization resource, and use the other functions in sqlalchemy_oso_cloud.orm to specify how properties correspond to facts. We use your model’s .id property as the resource ID, and use the model’s class name as the type.

sqlalchemy_oso_cloud.orm function	corresponding sqlalchemy.orm function	fact configured
relation	relationship	has_relation
remote_relation	mapped_column	has_relation
attribute	mapped_column	has_<name> (is_<name> if boolean)

For example, the following SQLAlchemy models configure the facts:

• has_relation(Folder:_, "organization", Organization:_)
• is_public(Folder:_)
• has_relation(Document:_, "folder", Folder:_)
• has_status(Document:_, String:_)

import sqlalchemy_oso_cloud.orm as oso
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Folder(Base, oso.Resource):
    __tablename__ = "folder"
    id: Mapped[int] = mapped_column(primary_key=True)
    organization_id: Mapped[int] = oso.remote_relation(remote_resource_name="Organization")
    name: Mapped[str]
    is_public: Mapped[bool] = oso.attribute(default=False)
    documents: Mapped[List["Document"]] = relationship(back_populates="folder")

class Document(Base, oso.Resource):
    __tablename__ = "document"
    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str]
    folder_id: Mapped[Optional[int]] = mapped_column(ForeignKey("folder.id"))
    folder: Mapped["Folder"] = oso.relation(back_populates="documents")
    status: Mapped[str] = oso.attribute()

Need a fact that isn’t covered by our functions? Open an issue and let us know.

Initialization

Use sqlalchemy_oso_cloud.init to initialize the integration. The function takes your SQLAlchemy model registry as its first argument, followed by any additional arguments you want to pass through when instantiating the Oso client.

import sqlalchemy_oso_cloud
from .models import Base

sqlalchemy_oso_cloud.init(
  Base.registry,
  url=os.environ["OSO_URL"],
  api_key=os.environ["OSO_AUTH"],
)

You can access the underlying Oso API client by calling sqlalchemy_oso_cloud.get_oso(). This returns the same client instance that was created during initialization.

from oso_cloud import Value
from sqlalchemy_oso_cloud import get_oso

oso = get_oso()
oso.insert(("has_role", Value("User", "alice"), "superadmin"))

Usage

The simplest way to use the SQLAlchemy integration is via sqlalchemy_oso_cloud.select, our drop-in replacement for SQLAlchemy’s built-in select function. It adds a method .authorized(user, permission), which filters the results of the query to only include resources the user has permission for.

from sqlalchemy_oso_cloud import select
from .session import Session
from .models import Document
from .authentication import get_current_user

with Session() as session:
    user = get_current_user()
    stmt = select(Document)
      .order_by(Document.created_at)
      .authorized(user, "read")
      .limit(10)
    documents = session.execute(stmt).scalars().all()

Using the legacy Query API

For applications using SQLAlchemy’s legacy Query API we provide sqlalchemy_oso_cloud.Session, a drop-in replacement for SQLAlchemy’s Session class. Its .query(...) method provides queries extended with the .authorized(...) method.

from sqlalchemy_oso_cloud import Session
from .engine import engine
from .models import Document
from .authentication import get_current_user

with Session(engine) as session:
    user = get_current_user()
    documents = session.query(Document)
      .order_by(Document.created_at)
      .authorized(user, "read")
      .limit(10)
      .all()

sqlalchemy_oso_cloud.Session is simply a convenience wrapper that initializes a sqlalchemy.orm.Session with our custom sqlalchemy_oso_cloud.Query class. If you can’t use our Session class, you can extend your own sessions:

from sqlalchemy_oso_cloud import Query
from sqlalchemy.orm import sessionmaker
from .db import engine

Session = sessionmaker(bind=engine, query_cls=Query)

Maximum Vanilla

sqlalchemy_oso_cloud.authorized lets you add authorization to any SQLAlchemy Select or Query via the existing .options(...) method.

from sqlalchemy_oso_cloud import authorized
from sqlalchemy import select
from .models import Document
from .session import Session

with Session() as session:
    user = get_current_user()
    stmt = select(Document)
      .order_by(Document.created_at)
      .options(authorized(user, "read", Document))
      .limit(10)
    documents = session.execute(stmt).scalars().all()

Using with Flask

sqlalchemy_oso_cloud.select is compatible with Flask-SQLAlchemy.

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy_oso_cloud import select
from sqlalchemy_oso_cloud.orm import Resource
from .authentication import get_current_user

db = SQLAlchemy()

class Document(db.Model, Resource):
  ...

with db.session() as session:
    user = get_current_user()
    stmt = select(Document)
      .order_by(Document.created_at)
      .authorized(user, "read")
      .limit(10)
    documents = session.execute(stmt).scalars().all()

If you use Flask-SQLAlchemy with the legacy Query API, initialize SQLAlchemy with our custom Query class to get the .authorized(...) method:

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy_oso_cloud import Query
from sqlalchemy_oso_cloud.orm import Resource
from .authentication import get_current_user

db = SQLAlchemy(query_class=Query)

class Document(db.Model, Resource):
  ...

with db.session() as session:
    user = get_current_user()
    documents = session.query(Document)
      .order_by(Document.created_at)
      .authorized(user, "read")
      .limit(10)
      .all()

Using with SQLModel

sqlalchemy_oso_cloud.select is compatible with SQLModel.

from sqlmodel import SQLModel, Session
from sqlalchemy_oso_cloud import select
from sqlalchemy_oso_cloud.orm import Resource
from .authentication import get_current_user
from .engine import engine

class Document(SQLModel, Resource):
  ...

with Session(engine) as session:
    user = get_current_user()
    stmt = select(Document)
      .order_by(Document.created_at)
      .authorized(user, "read")
      .limit(10)
    documents = session.exec(stmt).all()

If you use SQLModel with the legacy Query API, you can either use our Session class (which is compatible with SQLModel), or initialize SQLModel’s Session with our custom Query class to get the .authorized(...) method:

from sqlmodel import SQLModel, Session
from sqlalchemy_oso_cloud import Query
from sqlalchemy_oso_cloud.orm import Resource
from .authentication import get_current_user
from .engine import engine

class Document(SQLModel, Resource):
  ...

with Session(engine, query_cls=Query) as session:
    user = get_current_user()
    documents = session.query(Document)
      .order_by(Document.created_at)
      .authorized(user, "read")
      .limit(10)
      .all()

Reference

• GitHub repository
• API reference