Building a Multi-Tenant App with FastAPI, SQLModel, and PropelAuth
Building a multi-tenant (sometimes also known as B2B) product can be a trickier than building for consumers. Instead of handling users as individuals, you generally have to handle them working together in groups or teams.
In this tutorial, we’ll walk you through how to create a multi-tenant link shortening product using FastAPI (a popular Python web framework), SQLModel (a library for interacting with SQL databases in Python) and PropelAuth (a B2B/multi-tenant authentication provider).
We’ll start by setting up SQLModel so we can interact with the database.
Setting up the DB with SQLModel
SQLModel is the library we’ll use to interact with our database. It allows us to write code like this:
from sqlmodel import SQLModel, Field
class ExampleTable(SQLModel, table=True):
id: str = Field(primary_key=True)
some_field: str
which represents a SQL table with 2 columns (id and some_field) where the id is the primary key. It also makes querying, inserting, and most of our interactions with the DB simpler.
But before we get into that, we need to write some boilerplate code to set it up.
Initializing the DB Engine
First, we’ll need something that manages our connections to the DB. In SQLModel, this is called an engine. There should only ever be one engine across your entire application.
To create it, we'll make a new file db/db.py and paste in this code:
from sqlmodel import create_engine
engine = create_engine("sqlite:///database.db")
We’re using sqlite because, you know, this is a tutorial, but you can use other databases like Postgres as well.
While the engine is responsible for managing connections, querying/inserting data uses a different abstraction called a Session.
Querying Data with Sessions
With SQLModel, a simple “select all” SQL query looks like this:
with Session(engine) as session:
select_statement = select(ExampleTable)
return session.exec(select_statement).all()
We’ll typically want just one session per request, so instead of constructing the session manually in each of our routes, we can use dependency injection. All we have to do is add the following to db/db.py:
def get_session():
with Session(engine) as session:
yield session
In our future routes, we can then just inject the session like so:
@app.get("/session-example")
def session_example(session: Session = Depends(get_session)):
select_statement = select(ExampleTable)
return session.exec(select_statement).all()
We’ll see this again shortly, but for now, let’s set up some models.
Adding the Link model
As a reminder, our multi-tenant app is for link shortening. Each link will be tied to an “Organization” and only the members of that organization should be able to view/add/delete them. We’ll start by defining our table in a new file db/link.py
from sqlmodel import SQLModel, Field
class Link(SQLModel, table=True):
org_id: str = Field(primary_key=True)
id: str = Field(primary_key=True)
url: str
creator_user_id: str
We can then define a few helper functions for querying that table:
from sqlmodel import SQLModel, Field, Session, select
from nanoid import generate
# ... the model
def fetch_link(session: Session, org_id: str, link_id: str):
select_query = select(Link).where(Link.org_id == org_id, Link.id == link_id)
return session.exec(select_query).first()
def fetch_all_links(session: Session, org_id: str):
select_query = select(Link).where(Link.org_id == org_id)
return session.exec(select_query).all()
def save_link(session: Session, org_id: str, user_id: str, url: str):
link = Link(org_id=org_id, id=generate(), url=url, creator_user_id=user_id)
session.add(link)
session.commit()
session.refresh(link)
return link
The syntax should read similar to SQL itself. We’re using a Python port of nanoid to generate our IDs. There’s only one thing missing… how do we actually create the table?
Automatically creating tables
Creating the table is actually pretty easy, we just need to call SQLModel.metadata.create_all(engine)
The one caveat is we need to make sure all models are imported before we call that. To be safe, we’ll import all our models in our db/db.py file, even though we don’t use them there:
from . import link # noqa # pylint: disable=unused-import
Lastly, we need to actually call create_all. While you can call it directly, we can instead use FastAPI’s lifespan to call it before the server starts up:
from contextlib import asynccontextmanager
# ... imports, create engine, get_session
@asynccontextmanager
async def lifespan(_: FastAPI):
SQLModel.metadata.create_all(engine)
yield
Let’s now create our main.py, create our app, and hook up the lifespan:
from fastapi import FastAPI
app = FastAPI(lifespan=lifespan)
and now when you go to start the server with uvicorn main:app --reload, the tables will be automatically created.
Quick Recap
We’ve now successfully set up FastAPI to interact with a database.
We can define our models and the corresponding DB tables will be automatically created on server startup.
We have an ORM that we can use to query the DB.
And, maybe most important of all, we can dependency inject the session into our routes so we can interact with the DB without worrying about any of this code that we’ve written.
Creating our Routes
Setting up Authentication with PropelAuth
PropelAuth is an authentication service that specializes in multi-tenant products. They provide everything we need right out of the box, including UIs and APIs for signup, login, and account management, organization management, invite flows and more.
To get started, go to the PropelAuth Dashboard and create a project:
And now we are done. We can configure their look and feel and different options like how long users should stay logged in for, but let’s skip ahead and hook up our FastAPI backend.
Creating protected routes
Let’s create a new file called auth.py where we will initialize the propelauth-fastapi library:
from propelauth_fastapi import init_auth
auth = init_auth("AUTH_URL", "API_KEY")
You can find both your AUTH_URL and API_KEY in the Backend Integration section of the dashboard.
There’s a lot to discover on the auth object, but most important for us is going to be require_user. This is a dependency that will verify that the request was made by a valid user, otherwise it will return a 401 Unauthorized error.
We’ll start by making a route for submitting URLs in our main.py:
from pydantic import BaseModel
# ... imports and app creation
# We don't use the same Link class that the DB uses
# because we're going to want to fill in the user/org/id
# information ourselves
class LinkIn(BaseModel):
url: str
@app.post("/link")
def create_short_link(*,
session: Session = Depends(get_session),
link: LinkIn):
user_id = # TODO get user id?
org_id = # TODO get org id?
return save_link(session, org_id, user_id, link.url)
To get the user_id, we can use our require_user dependency:
@app.post("/link")
def create_short_link(*,
session: Session = Depends(get_session),
user: User = Depends(auth.require_user),
link: LinkIn):
org_id = # TODO get org id?
return save_link(session, org, user.user_id, link.url)
In PropelAuth, by default, users can be in more than one organization. While you can configure this, we recommend that the API explicitly take in the organization the user is creating the link for:
@app.post("/{org_id}/link")
def create_short_link(*,
session: Session = Depends(get_session),
user: User = Depends(auth.require_user),
org_id: str,
link: LinkIn):
# Make sure the user is in the specified organization
org = auth.require_org_member(user, org_id)
if not org:
raise HTTPException(status_code=404, detail="Not found")
return save_link(session, org.org_id, user, link.url)
We’re going to be doing this org_member check a few times, so let’s pull it out into it’s own dependency as well:
def require_org_member(*, user: User = Depends(auth.require_user), org_id: str):
org = auth.require_org_member(user, org_id)
if not org:
raise HTTPException(status_code=404, detail="Not found")
return org
@app.post("/{org_id}/link")
def create_short_link(*,
session: Session = Depends(get_session),
org: OrgMemberInfo = Depends(require_org_member),
user: User = Depends(auth.require_user),
link: LinkIn):
return save_link(session, org, user, link.url)
And this final route will:
- Make sure the user is logged in
- Make sure the user is in the specified organization
- Save the URL and attribute it to that user/organization
Building on this, the next two routes are pretty straightforward:
@app.get("/{org_id}/link")
def get_all_links(*,
session: Session = Depends(get_session),
org: OrgMemberInfo = Depends(require_org_member)):
return fetch_all_links(session, org.org_id)
@app.get("/{org_id}/{slug}")
def redirect_to_url(*,
session: Session = Depends(get_session),
org: OrgMemberInfo = Depends(require_org_member),
slug: str):
link = fetch_link(session, org, slug)
if not link:
raise HTTPException(status_code=404, detail="Not found")
return RedirectResponse(link.url)
Interestingly, both of these routes don’t actually care about who you are, they only care about which organization you are in. Thankfully, all that boilerplate is done by our require_org_member dependency.
Using Roles (RBAC) for our Delete Endpoint
We’re almost there, we just have one more route to make: delete. We first have to make a product decision - who can delete a given link?
In PropelAuth, you can configure your roles & permissions. The defaults are to have an Owner, Admin, and Member, like so:
We’ll leave these defaults alone, and for our route, let’s say that Owners and Admins can delete any link, whereas Members can only delete their own links. The top level route is straightforward:
@app.delete("/{org_id}/link/{slug}")
def delete_link(*,
session: Session = Depends(get_session),
org: OrgMemberInfo = Depends(require_org_member),
user: User = Depends(auth.require_user),
slug: str):
link = check_permission_and_delete_link(session, org, user, slug)
if not link:
raise HTTPException(status_code=404, detail="Not found")
return link
And now let’s go back to db/link.py and add our new function:
from propelauth_py.user import OrgMemberInfo, User
# ... other imports, model, and db functions
def check_permission_and_delete_link(session: Session, org: OrgMemberInfo, user: User, link_id: str):
link = fetch_link(session, org.org_id, link_id)
if not link:
return None
# Owners/Admins can delete any link,
# but users can only delete their own links
if org.user_is_at_least_role("Admin") or link.creator_user_id == user.user_id:
session.delete(link)
session.commit()
return link
return None
The OrgMemberInfo class has a few helper functions like user_is_at_least_role which we use to check if the user is an Admin or Owner.
Aside: How are users added to organizations? How do they get roles?
You may have noticed that we haven’t written any code around the meta of organization management - creation, adding and removing users, etc. That’s because the PropelAuth hosted pages will take care of all of that for us. Our frontend will handle displaying the UIs to the user, and they’ll be able to self-serve. If we need to do any management on our end, the PropelAuth dashboard will allow us to do that without writing any code (or they have APIs if you want more control).
Even better, if we start selling to larger organizations and they ask for SAML (a.k.a. enterprise SSO, Login via Okta, and many more ways to say the same thing), our code still doesn’t change. All we have to do is mark their organization as “Can setup SAML,” and the customer can setup a SAML connection without us changing our code at all.
Testing your API
FastAPI OpenAPI UI
One of my personal favorite features of FastAPI is it’s built-in OpenAPI support. If you go to http://localhost:8000/docs, you’ll be greeted with this OpenAPI page:
It includes all our APIs, and notably also includes an “Authorize” button so we can make authorized requests.
Before we do that though, let’s see what happens if we don’t specify any auth information:
Unsurprisingly, we get a 401 Unauthorized error.
Let’s go back and click that Authorize button:
It’s asking for a token for one of our users. When we have a frontend, this is easy because the frontend libraries all have ways to get access tokens.
However, PropelAuth also provides ways to generate access tokens without a frontend so you can test your backend independently. We’ll use the following curl command:
curl -H "Content-Type: application/json"
-H "Authorization: Bearer {PROPELAUTH_API_KEY}"
-X POST
-d '{"user_id": "{USER_ID}", duration_in_minutes: 1440}'
"{AUTH_URL}/api/backend/v1/access_token"
Note that you’ll need to fill that in with the same API Key and Auth URL we used before. You also need to pass in a user_id which you can get by going to Users, selecting one, and clicking Copy User ID.
Take the access token that you get back from that API call and paste it into the Authorize box, and now we are ready to test.
If you call the “Get all links” endpoint with an organization the user is in, you’ll get [] because we haven’t made any links yet.
If you call the “Create short link” endpoint with an organization the user is in, and a URL, you can see:
That the creator_user_id is set to us, the org_id is set to the one we specified, and an ID is automatically generated.
To be safe, what happens if we specify an org_id that the user is NOT in?
We get a 403 Forbidden exception.
There’s more APIs that you can test here, but this should cover the basics. We can authenticate and then scope all our actions to a single organization that the user must be in.
Testing with a real frontend
Testing the backend in isolation is really helpful for making sure the APIs work before you have a frontend. Once you do have a frontend, you’ll just want to make sure that the request is formatted correctly. We’ve written a quick guide on that here which shows you how to get the access token on the frontend and format your requests.
Summary
Now that your multi-tenant link shortener is complete, you should be able to use FastAPI, SQLModel and PropelAuth to get your own product up and running.
If you’d like to read more about any of these technologies, here are links to their documentation: