SQLAlchemy Integration

Our open-source (opens in a new tab) integration for SQLAlchemy (opens in a new tab) 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 (opens in a new tab) 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 (opens in a new tab).


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 (opens in a new tab).


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

💡

Have an unsupported use case? Open an issue (opens in a new tab) to let us know.

Migrating to v2