Tutorial: Using Cerbos with SQLAlchemy

from sqlalchemy import Column, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "user"

    id = Column(String, primary_key=True)
    username = Column(String(255))
    email = Column(String(255))
    # ...

class User(Base):
    __tablename__ = "user"

    # ...
    contacts = relationship("Contact", back_populates="owner")

class Contact(Base):
    __tablename__ = "contact"

    id = Column(String, primary_key=True)
    # ...
    owner_id = Column(String, ForeignKey("user.id"))
    owner = relationship("User", back_populates="contacts", lazy="joined")

from sqlalchemy import select

# Session is a SQLAlchemy sessionmaker instance
with Session() as s:
    user = s.scalars(select(User).where(User.username == "gandalf")).first()
    user.email  # "greybeard99@midearth.com"

    # Note, in order to reference contacts with a `lazy` loading pattern, the
    # attribute lookup needs to occur in the context of a session - hence it's
    # in the Session() context manager scope.
    contact = user.contacts[0]
    contact.user_id == user.id  # True

from sqlalchemy import create_engine
from sqlalchemy.pool import StaticPool

engine = create_engine(
    "sqlite://",  # the absence of a specified URL infers a `:memory:` database (e.g. no disk persistence)
    connect_args={"check_same_thread": False},  # in FastAPI, when using sync (def) functions, more than one thread could interact with the database
                                                # for the same request, so we need to make SQLite know that it should allow that
    poolclass=StaticPool,  # Use a static pool to persist state with an in memory instance of sqlite
)

Base.metadata.create_all(engine)

with Session() as s:
    coca_cola = Company(name="Coca Cola")
    s.add(coca_cola)
    s.commit()

    john = User(
        name="John",
        username="john",
        email="john@cerbos.demo",
        role="user",
        department="Sales",
    )
    s.add(john)
    s.commit()

    s.add(Contact(
            first_name="Nick",
            last_name="Smyth",
            marketing_opt_in=True,
            is_active=True,
            owner=john,
            company=coca_cola,
    ))
    s.commit()

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

def get_principal(credentials: HTTPBasicCredentials = Depends(security)) -> Principal:
    username = credentials.username

    with Session() as s:
        # retrieve `user` from the DB to access the attributes
        user = s.scalars(select(User).where(User.username == username)).first()
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="User not found",
            )

    return Principal(user.id, roles={user.role}, attr={"department": user.department})

@app.get("/contacts")
def get_contacts(p: Principal = Depends(get_principal)):
    # do something with the principal "p"

def get_db_contact(contact_id: str) -> Contact:
    with Session() as s:
        contact = s.scalars(select(Contact).where(Contact.id == contact_id)).first()
        if contact is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Contact not found",
            )
    return contact

def get_resource_from_contact(
    db_contact: Contact = Depends(get_db_contact),
) -> Resource:
    return Resource(
        id=db_contact.id,
        kind="contact",
        attr=jsonable_encoder(
            {n.name: getattr(db_contact, n.name) for n in Contact.__table__.c}
        ),
    )

@app.delete("/contacts/{contact_id}")
def delete_contact(
    r: Resource = Depends(get_resource_from_contact),
    p: Principal = Depends(get_principal),
):
    # do something with the resource

@app.get("/contacts/{contact_id}")
def get_contact(
    db_contact: Contact = Depends(get_db_contact),
    p: Principal = Depends(get_principal)
):
    # optionally, call the dependable direct to retrieve the resource from the db Contact instance
    resource = get_resource_from_contact(db_contact)

from pydantic import BaseModel

class ContactSchema(BaseModel):
    first_name: str
    last_name: str
    owner_id: str
    company_id: str
    is_active: bool = False
    marketing_opt_in: bool = False

    class Config:
        # tell the Pydantic model to read the data even if it is not a dict, but an ORM model
        # (or any other arbitrary object with attributes)
        orm_mode = True

@app.post("/contacts/new")
def create_contact(
    contact_schema: ContactSchema, p: Principal = Depends(get_principal)
):
    with CerbosClient(host="http://localhost:3592") as c:
        if not c.is_allowed(
            "create",
            p,
            Resource(
                id="new",
                kind="contact",
            ),
        ):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="Unauthorized"
            )

    db_contact = Contact(**contact_schema.dict())
    with Session() as s:
        s.add(db_contact)
        s.commit()
        s.refresh(db_contact)

    return {"result": "Created contact", "contact": db_contact}

@app.get("/contacts/{contact_id}")
def get_contact(
    db_contact: Contact = Depends(get_db_contact), p: Principal = Depends(get_principal)
):
    r = get_resource_from_contact(db_contact)

    with CerbosClient(host="http://localhost:3592") as c:
        if not c.is_allowed("read", p, r):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="Unauthorized"
            )

    return db_contact

@app.get("/contacts")
def get_contacts(p: Principal = Depends(get_principal)):
    with CerbosClient(host="http://localhost:3592") as c:
        rd = ResourceDesc("contact")

        # Get the query plan for "read" action
        plan = c.plan_resources("read", p, rd)
        print(json.dumps(plan.to_dict(), sort_keys=False, indent=4))

    query = get_query(
        plan,
        Contact,
        {
            "request.resource.attr.owner_id": User.id,
            "request.resource.attr.department": User.department,
            "request.resource.attr.is_active": Contact.is_active,
            "request.resource.attr.marketing_opt_in": Contact.marketing_opt_in,
        },
        [(User, Contact.owner_id == User.id)],
    )

    # Optionally reduce the returned columns (`with_only_columns` returns a new `select`)
    # NOTE: this is wise to do as standard, to avoid implicit joins generated by sqla `relationship()` usage, if present
    query = query.with_only_columns(
        Contact.id,
        Contact.first_name,
        Contact.last_name,
        Contact.is_active,
        Contact.marketing_opt_in,
    )
    print(query.compile(compile_kwargs={"literal_binds": True}))

    # ...
    with Session() as s:
        rows = s.execute(query).fetchall()

    return rows

git clone git@github.com:cerbos/python-sqlalchemy-cerbos.git
cd python-sqlalchemy-cerbos

cd cerbos
./start.sh

# from project root
pdm install

pdm run demo