Building a Multi-Tenant App with FastAPI, SQLModel, and PropelAuth

victoria_propel

Victoria

Posted on December 6, 2023

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
Enter fullscreen mode Exit fullscreen mode

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")
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

Let’s now create our main.py, create our app, and hook up the lifespan:

from fastapi import FastAPI

app = FastAPI(lifespan=lifespan)
Enter fullscreen mode Exit fullscreen mode

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:

Create Project screen in PropelAuth

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")
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode
@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)
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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:

List of roles within PropelAuth

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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:

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:

No auth information

Unsurprisingly, we get a 401 Unauthorized error.

Let’s go back and click that Authorize button:

Authorize UI

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" 
Enter fullscreen mode Exit fullscreen mode

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.

Users page in PropelAuth

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:

Create short link UI

Response body

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?

403 example

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:

FastAPI

SQLModel

PropelAuth

💖 💪 🙅 🚩
victoria_propel
Victoria

Posted on December 6, 2023

Join Our Newsletter. No Spam, Only the good stuff.

Sign up to receive the latest update from our blog.

Related