---
title: "SQLAlchemy authorization: How to implement RBAC & ABAC"
description: "Read our guide on how to set up SQLAlchemy with Cerbos for easy authorization or user permission aware data filtering or fetching in your application."
author: "Sam Lock"
date: "2024-03-20T01:00:00.000Z"
canonical: "https://www.cerbos.dev/blog/sqlalchemy-authorization"
image: "https://stylish-appliance-1c1cc1c30d.media.strapiapp.com/cover_image_3d7c9a67eb.png"
tags: ["documentation","guide","integration"]
source: "https://www.cerbos.dev/blog/sqlalchemy-authorization"
---
# SQLAlchemy authorization: How to implement RBAC & ABAC
If you maintain an application that handles any _state_ at all, it's likely that you've had to figure out how to both store that state, as well as how to load it into the application layer and act on it in any which way your business logic requires.
Perhaps, in your case, a lot of the computational "heavy lifting" is done by the database, and the application is just an abstraction layer where you write your database queries. Or maybe on the contrary, the database is just a basic store which provides the data for the application to manage all of the tricky logic itself.
Regardless, there's _many_ ways to build an application (as the common idiom doesn't go). Application design is a vast and complex process, but one thing we can do to make that process more manageable is to use tools that take a lot of the implementation complexity away...
#### Enter SQLAlchemy
[SQLAlchemy](https://www.sqlalchemy.org/) has established itself as one of the standards in database abstraction layers in the Python world. It offers two distinct ways of communicating with the DB; via it's lower-level`Core` SQL abstraction toolkit, or via it's`ORM` component, which extends `Core` to offer some convenient, higher-level abstractions.
#### What we're building
In this run-through, we'll be building an application that manages a "Contact directory", enabling users to keep track of their contacts, along with useful information such as employment information (current company etc).
We're going to explore how to model our data, map it to[Cerbos](https://cerbos.dev) entities, and interact with it in a clean, efficient and reusable way.
We'll be building a Python[FastAPI](https://fastapi.tiangolo.com/) server and securing it using the following Cerbos APIs:
* `CheckResources`: e.g. can `User X` from the Sales department access `Contact Y`?
* `PlanResources`: e.g. which contacts can `User X` from the Marketing department access?
The full source code for this demo can be found in our repo [here](https://github.com/cerbos/python-sqlalchemy-cerbos).
[](https://cta-service-cms2.hubspot.com/web-interactives/public/v1/track/redirect?encryptedPayload=AVxigLKntt8fdQvjLtfXzPmwXqWXdClso1u%2FMwtyDq8%2Fjwp8XQC5MIuRht4cOw6DsxFsvcPCiFNbn57jrpHSb0a2cdTgR6BSX1w4YPJ42R0PvmaKh79f4FlDHlYmC6lMtFt7p8U4I17YrAI3dtqTAus%2BEHAApNn1h76t%2Bwyl2sc%3D&webInteractiveContentId=168994186087&portalId=20289770)
## Prerequisites
* Python 3.10
* [SQLAlchemy](https://docs.sqlalchemy.org/en/14/) 1.4 / 2.0
* [Docker](https://www.docker.com/products/docker-desktop/) running locally.
## The database
### Setting up our models
We have the following entities within our application:
* `User`: the person interacting with the application
* `Contact`: a person within a `User`'s directory (a`User` can have many `Contacts`)
* `Company`: the company that a `Contact` is currently employed with (a `Company` can have many `Contacts`)
In order to persist and manage these models, we need to be able to represent them in code in a way that can be mapped to our database layer. This is where SQLAlchemy comes in.
SQLAlchemy allows us to represent our relational database tables as classes, with attributes representing the columns of those tables. An object instance of one of these classes will represent a single row in the table. An example is shown below:
```python
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))
# ...
```
It also allows us to go a step further, and model relationships between these tables (via variations of table joins). In our case, we want to be able to model the one-to-many relationships mentioned above:
```python
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")
```
You can see how we relate the two tables via the`relationship` function. In setting a`relationship` field on _each_ linked class, we establish a bidirectional relationship between the objects (with the "reverse" side being a many-to-one). In this particular case, the `ForeignKey` placed on the child table infers the many-to-one side, and as such, allows for child table objects to reference the parent via`child.owner`/`child.owner_id`. The`lazy="joined"` parameter indicates to SQLAlchemy that we'd like to lazily load the related object at attribute access time.
The full table definitions can be found in [this module](https://github.com/cerbos/python-sqlalchemy-cerbos/blob/main/app/models.py).
You can see how SQLAlchemy ORM entity objects can then be used to reference each another in code:
```python
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
```
Check out the excellent [SQLAlchemy documentation](https://docs.sqlalchemy.org/en/14/orm/relationships.html) for more information on relationships.
### Connecting to our database
SQLAlchemy is a wonderful abstraction layer between Python and a whole array of different relational databases. By specifying the "dialect" when connecting to a DB engine, we tell it which relational database it is connecting to.
For our demo, we’ll be setting up a simple, ephemeral SQLite instance. We won’t even persist it to disk; each time the application is started, it’ll build the DB in memory and populate it with a migration script.
We create the engine like so:
```python
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
)
```
### Tables and metadata
> To start using the SQLAlchemy Expression Language, we will want to have`Table` objects constructed that represent all of the database tables we are interested in working with. Each `Table` may be**declared**, meaning we explicitly spell out in source code what the table looks like, or may be **reflected**, which means we generate the object based on what’s already present in a particular database.
>
> Whether we will declare or reflect our tables, we start out with a collection that will be where we place our tables known as the MetaData object. This object is essentially a facade around a Python dictionary that stores a series of Table objects keyed to their string name.
Our classes above inherit from a base class generated from a call to`declarative_base()`. This "declarative" method allows us to declare user-defined classes and `Table` metadata at once. Each time a class inherits from this `Base` class, it is added to this collection, or `registry`. The following call will generate the database tables from the metadata:
```python
Base.metadata.create_all(engine)
```
### Populating the database
We can then generate a `session` from our`engine` instance, and use it to populate our newly generated tables:
```python
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()
```
You can see in the example above (in the `Contact` definition) how we can define relationships by referencing instances of the table classes.
Again, the full source code for this section can be found [here](https://github.com/cerbos/python-sqlalchemy-cerbos/blob/main/app/models.py).
## The API
We now have a database which can be declared and populated on demand, and models which allow us to interact with it. The next step is to build an API layer to expose the data, and to secure the endpoints and resources with Cerbos.
We'll be creating our server with FastAPI. The source code for this section can be found [here](https://github.com/cerbos/python-sqlalchemy-cerbos/blob/main/main.py).
### Dependency injection with FastAPI dependables
FastAPI allows you to define callables called "dependables", which are functions that take all of the same arguments as a "path operation function" and return whatever we might require for the handler. The`Depends(fn)` class takes the callable and on execution will return the default argument, if required. We define a few dependables which we can use across our endpoints.
Firstly, one to retrieve the cerbos Principal instance from the username (which in itself is retrieved via the FastAPI provided`HTTPBasic` dependable):
```python
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})
```
This can then be used on all endpoints to authenticate the user, and then assert that the user exists in the database:
```python
@app.get("/contacts")
def get_contacts(p: Principal = Depends(get_principal)):
# do something with the principal "p"
```
We then create a dependable which attempts to retrieve the`Contact` from the database based on a path parameter in the URL:`contact_id`:
```python
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
```
This in turn can then be nested in another dependable which attempts to return the `Contact` Resource instance:
```python
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}
),
)
```
These can then be used to attempt to retrieve a Cerbos `Resource`, or a SQLAlchemy `Contact` instance respectively, on routes which include the `contact_id` path parameter:
```python
@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)
```
### Defining our API schema
Some of our routes will require specific parameters in the payload in order to carry out the given request. For example, we might need routes for creating or updating new or existing `Contacts`.
FastAPI provides a nice interface to enforce request schema via "Pydantic" models:
```python
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
```
Once we've defined these schema models, we can use them in the FastAPI routes:
```python
@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}
```
FastAPI will automatically validate the input payload to ensure required fields are present, and types are correct (as well as other optional checks). We can then use the schema model attributes to generate SQLAlchemy models as you can see above.
The schema for our demo can be found in [this module](https://github.com/cerbos/python-sqlalchemy-cerbos/blob/main/app/schemas.py).
### Protecting the routes
Now we have our dependables and API schema defined, we can start to define the routes and secure them using Cerbos.
We can make granular checks against specific `principal:resource:action` mappings using Cerbos'`CheckResources` API (via the `is_allowed` method):
```python
@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
```
However, sometimes, we want to establish which resources a Principal has access to. To do this, we can use the `PlanResources` API.
### The query planner
If we provide Cerbos with a `Principal` and a description of the resource they're trying to access (`ResourceDesc`), we can ask it for a query plan.
The `PlanResources` call returns one of the following:
* `KIND_ALWAYS_ALLOWED`
* `KIND_ALWAYS_DENIED`
* `KIND_CONDITIONAL`
In the final case, it'll also return an abstract syntax tree (AST) of the condition that must be satisfied to allow the action:
```python
@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))
```
Cerbos provides a [SQLAlchemy adapter library](https://github.com/cerbos/query-plan-adapters/tree/main/sqlalchemy) with an API that takes the query plan response, and uses it to generate a SQLAlchemy query object. Continuing below:
```python
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}))
```
The provided `get_query` function accepts the following parameters, respectively:
1. query plan
2. a primary SQLAlchemy `Table` or ORM`DeclarativeMeta` type (the `FROM table` part of the resulting query)
3. the "attribute map" - responsible for mapping the Cerbos resource attribute strings to the associated SQLAlchemy columns (type `Column` or ORM `InstrumentedAttribute`)
4. OPTIONAL: list of explicit table joins - required only if more than one table specified in primary table + attribute map
It returns a SQLAlchemy `Selectable`, which can be further extended/reduced, and then used to query the database:
```python
# ...
with Session() as s:
rows = s.execute(query).fetchall()
return rows
```
### Run the server
Now we understand how everything works, let's fire up the server and the Cerbos PDP, and test it out.
Clone the repo:
```sh
git clone git@github.com:cerbos/python-sqlalchemy-cerbos.git
cd python-sqlalchemy-cerbos
```
Start up the Cerbos PDP instance docker container:
```sh
cd cerbos
./start.sh
```
Install Python dependencies:
```sh
# from project root
pdm install
```
Start the FastAPI dev server:
```sh
pdm run demo
```
### Example requests
#### Get all permitted contacts
```sh
curl http://john@localhost:8000/contacts
```
#### Get a single contact
Sales user, contact owned => `200 OK`
```sh
curl -i http://john@localhost:8000/contacts/1
```
Sales user, contact not owned or active => `403 Forbidden`
```sh
curl -i http://john@localhost:8000/contacts/4
```
#### Create a contact
Sales user => `200 OK`
```sh
curl -i http://john@localhost:8000/contacts/new \
-H 'Content-Type: application/json' \
-X POST \
-d '{"first_name": "frodo", "last_name": "baggins", "owner_id": "2", "company_id": "2"}'
```
Marketing user (e.g. `geri`) => `403 Forbidden`
#### Delete a contact
Contact owner => `200 OK`
```sh
curl -i http://john@localhost:8000/contacts/1 -X DELETE
```
Non-owner => `403 Forbidden`
```sh
curl -i http://john@localhost:8000/contacts/3 -X DELETE
```
---
## The benefits of using Cerbos for SQLAlchemy authorization
SQLAlchemy is an object relational mapper and Python toolkit that provides developers with the ability to take maximum advantage of the power of SQL by delivering highly efficient and effective database access and schema management. By using Cerbos for SQLAlchemy authorization you produce a variety of advantages and benefits for your application that include greater access control, enhanced scalability and, of course, a higher level of security. Here’s an overview of each consideration:
Greater security: One of the primary benefits of using Cerbos as your authorization system is the way it boosts application security. The Cerbos policy-driven model enables developers to create fine-grain access policies and implement them application wide with speed and ease. The degree of fine-grain control you can achieve is impressive and will allow you to effectively mitigate the risk of unauthorized access to the most sensitive files in your SQLAlchemy DB.
More dependable compliance: By choosing Cerbos to be your SQLAlchemy authorization gatekeeper you greatly enhance your compliance and auditing capabilities. Cerbos automatically generates comprehensive logs and insights regarding all access requests. The result is complete transparency into user activity. You will know exactly who requested and was granted access, when they requested access and what resources they accessed. Such a detailed audit trail will enable you to fulfil your compliance obligations and meet regulatory requirements.
Flexibility: SQLAlchemy is known for its flexibility in handling data models and various database systems which makes it a good match for the fine-grained access control you’ll enjoy with Cerbos. The centralized control provided by Cerbos enables you to quickly and easily define and update access control policies, meaning you can make changes on the fly and evolve your access control mechanisms as need dictates.
Scalability: No developer expects an SQLAlchemy application to remain a static object. Needs change, companies grow and applications need to grow with them. Cerbos accommodates increased data volumes and instances of user interaction without skipping a beat due to its centralized policy management system. In no time you can redefine access policies, update authorization requirements and add or remove attributes with application-wide implications.
## Conclusion
When you choose Cerbos to provide SQLAlchemy authorization you are choosing flexibility, scalability, enhanced compliance and greater security. Cerbos and SQLAlchemy are symbiotic applications that dovetail seamlessly with the end result being an SQLAlchemy application that is greater than the sum of its parts.
If you have any questions or feedback, or to chat to us and other like-minded technologists, please join our [Slack community](https://community.cerbos.dev)!
[](https://cta-service-cms2.hubspot.com/web-interactives/public/v1/track/redirect?encryptedPayload=AVxigLLcsDgffnFDNxbAukQV0SmAqw%2Bh1lAEElcTb3BQwSAIXmQ4WXfJgejBOSkzLZ7W7VbZmSrj7w25pmEjE1nho266aSJYw69kGnULNoeFa05RoQ7F7uWVzJwoPaT5KDyIEDJ3FnTp756bSD6DYLowFW3qMjOTWrrsWp4oc7yRzm%2Bvl5TdqcXJHVVu8Tlx&webInteractiveContentId=168996984756&portalId=20289770)
## FAQ
### Is SQLAlchemy a good database ORM?
SQLAlchemy is a versatile tool designed for interacting with a backend database. It's suitable for a range of tasks, from creating a frontend web application to managing a database through code. This flexibility makes it a go-to solution for various database-related projects, offering a streamlined way to work with data across different applications.
### How do I implement authorization in SQLAlchemy?
To implement authorization in SQLAlchemy, use Cerbos for policy-driven, fine-grained access control. First, set up SQLAlchemy models and use Cerbos to define access policies (e.g., for roles like "admin" or "user"). Use Cerbos’s CheckResources to verify if a user can access specific records and PlanResources to determine accessible resources based on policies. Integrate these checks within FastAPI endpoints to handle permissions, ensuring secure, scalable, and compliant access control across your app.
### What patterns can be used for SQLAlchemy access control?
For SQLAlchemy access control, these patterns could be used:
- Role-Based Access Control (RBAC) - Assigns permissions based on user roles like "admin" or "user."
- Attribute-Based Access Control (ABAC) - Grants access based on user attributes (e.g., department).
- Policy-Driven models - Use tools like Cerbos to define and manage centralized policies.
- Ownership checks - Restrict access based on data ownership, useful in multi-user applications.
Each pattern suits different access needs.
### How can I enforce permissions in SQLAlchemy models?
To enforce permissions in SQLAlchemy models, use a centralized policy or access control layer. Define access rules based on roles (RBAC), attributes (ABAC), or ownership checks. Integrate these checks within model methods or at query time using filters based on user roles or attributes. For complex needs, use an authorization tool that applies policy-based checks before accessing or modifying data, ensuring security and consistency across your app.