FastAPI

FastAPI in Prod: Handling DB migrations, auth, and more

Andrew Israel

• 9 min read
FastAPI in Prod: Handling DB migrations, auth, and more

FastAPI is a modern Python framework known for its speed, developer-friendly features, and (my personal favorite) automatic OpenAPI schema generation. It provides you with powerful features while still remaining fairly unopinionated (relative to something like Django).

In this post, we’ll create a full example backend that includes libraries/tools you’ll need to get ready to ship to production. Here’s a quick rundown of what we’ll use:

  • dbmate – A simple, language-agnostic approach to managing database migrations.
  • PugSQL – For interacting with your database. It lets you write plain SQL in separate files for clarity, while keeping query calls as straightforward Python functions.
  • PropelAuth – Offers out-of-the-box B2B authentication, so you can handle multi-tenant org checks and user roles without building it yourself.
  • Pydantic – Ensures your incoming data is valid (like checking if a string is a URL). It’s “the most widely used data validation library for Python” and integrates easily with FastAPI.
  • nanoid – A library for generating short, unique IDs (like NxWDZ-j95C).

Our example backend will be a multi-tenant “bookmark aggregator.” Users will be able to bookmark URLs and share them with their team. Let’s get started by setting up the database.

Setting up the Database with dbmate

No matter what language or framework you are using, you’ll need something to manage your database schema. dbmate offers a simple approach - write migrations directly in SQL ⇒ use a CLI to apply or revert migrations.

After installing dbmate, we can create our initial migration:

dbmate new create_bookmarks_table

You’ll see a file like 20250101235959_create_bookmarks_table.sql in a migrations folder. Let’s define our bookmarks table:

-- migrate:up
CREATE TABLE bookmarks (
  bookmark_id TEXT NOT NULL PRIMARY KEY,
  link TEXT NOT NULL
);

-- migrate:down
DROP TABLE bookmarks;

This allows us to define both the migration and how to revert it. Before we can run the migration, we’ll need to tell dbmate where our DB lives. For testing purposes, we’ll use SQLite, but you can also use Postgres, MySQL, or ClickHouse.

# .env
DATABASE_URL="sqlite:db/database.sqlite3"

Now we can run:

dbmate up

which will both create our database (since this is the first time we ran it) and run any pending migrations (in this case, just our one migration). As a nice bonus, dbmate will also create a schema.sql file so you don’t need to parse through a bunch of migrations to see the current schema in full.

Querying the DB with PugSQL

I’m a very big fan of libraries that let you write raw SQL, and in Python, PugSQL is a really great option. With PugSQL, you can write queries in .sql files and call them from your Python code.

As an example, we can create a file named db/queries/get_link.sql

-- :name get_link :scalar
SELECT link FROM bookmarks WHERE bookmark_id = :bookmark_id;
  • :name get_link means the function name in Python will be get_link
  • :scalar means we’re returning a single value (just a link)
  • :bookmark_id is a parameter that we will pass into our Python get_link function

Similarly, we will create another file db/queries/save_bookmark.sql

-- :name save_bookmark :insert
INSERT INTO bookmarks (bookmark_id, link) VALUES (:bookmark_id, :link);

which allows us to insert values into the bookmarks table.

The docs will show you more examples - like how you can use transactions or how the save_bookmark function already supports batched inserts.

Note that another good option here is SQLModel which will integrate directly with your Pydantic models (see later section on validation). I just really like writing SQL directly.

Now it’s time to start setting up FastAPI.

Putting it together in FastAPI

After installing our dependencies pip install pugsql "fastapi[standard]", we can now set up our application’s main.py:

import pugsql
from fastapi import FastAPI, HTTPException

app = FastAPI()

# Load our queries from the db/queries folder
queries = pugsql.module('db/queries')
queries.connect("sqlite:///db/database.sqlite3")

@app.get("/{bookmark_id}")
async def redirect_to_bookmark(bookmark_id: str):
    link = queries.get_link(bookmark_id=bookmark_id)
    if link is None:
        raise HTTPException(status_code=404, detail="Bookmark not found")
    return {"link": link}

All pretty straightforward so far! We have a generic route that will query the DB for links by their ID. If a link is found, we redirect the user, otherwise we 404.

The only problem is… we don’t have any links. But first, we need to create an endpoint that only accepts valid ones.

Validating requests with Pydantic

FastAPI supports Pydantic models for validating requests. In our case, we only need to take in one value - the URL.

Luckily, Pydantic has types specifically for URL validation, meaning our model is just:

from pydantic import BaseModel, HttpUrl

class Bookmark(BaseModel):
    link: HttpUrl

We can then hook this up directly:

@app.post("/bookmark")
async def create_bookmark(bookmark: Bookmark):
    bookmark_id = # TODO
    queries.save_bookmark(bookmark_id=bookmark_id, link=bookmark.link)
    return {"bookmark_id": bookmark_id}

This actually does a number of things for us:

Next, we’ll need to generate a unique bookmark_id. Normally, you might just use a uuid here, but if we want a slightly more “friendly” looking link, we can instead use a nanoid.

After running pip install nanoid, we have a choice to make. We can generate a standard 21 character nanoid, which will have collision probability similar to UUID v4 (in other words, you don’t have to worry about collisions):

from nanoid import generate

generate() # => NDzkGoTCdRcaRyt7GOepg

or we can make shorter IDs, but increase the chance of a collision:

generate(size=10) # => "IRFa-VaY2b"

If you do go with shorter IDs with more likely collusions, just make sure to catch the PugSQL exception and try again. In our case, we can just use 21 characters:

from nanoid import generate

@app.post("/bookmark")
async def create_bookmark(bookmark: Bookmark):
    bookmark_id = generate()
    queries.save_bookmark(bookmark_id=bookmark_id, link=bookmark.link)
    return {"bookmark_id": bookmark_id}

Testing our API

We can run our server with fastapi dev [main.py](<http://main.py/>) and then make a simple cURL request:

curl --data '{"link": "https://example.com"}' http://localhost:8000/bookmark

and we’ll get back something like:

{"bookmark_id": "NDzkGoTCdRcaRyt7GOepg"}

Notably, we can also navigate to http://localhost:8000/docs and use the OpenAPI UI to make this request instead of using cURL.

Afterwards, we can go to http://localhost:8000/NDzkGoTCdRcaRyt7GOepg and get back a response returning https://example.com.

Adding Authentication and Multi-Tenancy with PropelAuth

All of our routes so far have been unprotected - anyone can make requests to them. We’re first going to protect our routes to make sure only authenticated users can make requests, then we are going to add multi-tenancy so users can only create/view bookmarks in their tenant.

After signing up and creating a project, we can protect our route by adding one dependency:

# make sure to run `pip install propelauth_fastapi`
from propelauth_fastapi import init_auth, User
from fastapi import FastAPI, Depends

# these values can be found in the `Backend Integration` section of the dashboard
auth = init_auth("YOUR_AUTH_URL", "YOUR_API_KEY")

@app.post("/bookmark")
async def create_bookmark(
    bookmark: Bookmark, 
    current_user: User = Depends(auth.require_user)
):
    # only authenticated requests are now allowed
    bookmark_id = generate()
    queries.save_bookmark(bookmark_id=bookmark_id, link=bookmark.link)
    return {"bookmark_id": bookmark_id}

The signup, login, and account pages are managed for us (or you can build your own), and features like SSO, MFA, and session management can be enabled/configured in your dashboard.

But we are building a B2B application, which means users won’t interact with our product in isolation - they will use our product with other members of their organization (or team, company, tenant - there’s a lot of words for this).

Luckily, PropelAuth was built specifically for B2B use cases and has a first class concept of organizations as well as roles & permissions (RBAC). Our users can create an organization, invite their coworkers, and manage the members of the organization already, so all we need to do is check it their org:

@app.post("/{org_id}/bookmark")
async def create_bookmark(
    bookmark: Bookmark, 
    org_id: str,
    current_user: User = Depends(auth.require_user)
):
    # Make sure the user is in the organization
    if user.get_org(org_id) is None:
        raise HTTPException(status_code=401, detail="Not a member of this org")

    bookmark_id = generate()
    # TODO: bookmark should be scoped to the organization
    queries.save_bookmark(bookmark_id=bookmark_id, link=bookmark.link)
    return {"bookmark_id": bookmark_id}

Note that there are a few ways to specify an organization that the user is interacting with:

  • Path Parameter (e.g. https://api.example.com/{org_id}/bookmark) - this tends to be the easiest one to test locally and has the benefit of being explicit.
  • Embedded in the token (e.g. https://api.example.com/bookmark, but the frontend will mark a single “active” organization) - conceptually simple, a little harder to test without having a frontend set up.
  • Subdomain (e.g. https://{org_name}.example.com/bookmark) - this is harder to test locally. Since this is often done for vanity reasons, you can actually do this on the frontend (so the user sees https://{org_name}.example.com) but your API can use a path parameter instead.

We’ll stick with the path parameter for ease of local testing.

Running another DB migration

When we first created our DB schema, we didn’t have a concept of users or organizations. Let’s update our schema now to include that:

dbmate new add_user_and_org_ids

In the file that’s created:

-- migrate:up
ALTER TABLE bookmarks ADD COLUMN org_id TEXT NOT NULL;
ALTER TABLE bookmarks ADD COLUMN user_id TEXT NOT NULL;

-- migrate:down
ALTER TABLE bookmarks DROP COLUMN org_id;
ALTER TABLE bookmarks DROP COLUMN user_id;

Note that the NOT NULL constraint is only valid if you haven’t created any bookmarks yet. We can just clear the DB since we don’t want any unattributed rows. In this case, you may also want to re-create the whole table to change the primary key to (org_id, bookmark_id).

Then we can run it:

dbmate up

And finally, we’ll want to make sure we update our save_bookmark.sql and get_link.sql queries:

-- :name save_bookmark :insert
INSERT INTO bookmarks (bookmark_id, link, org_id, user_id) 
  VALUES (:bookmark_id, :link, :org_id, :user_id);

-- :name get_link :scalar
SELECT link FROM bookmarks 
  WHERE bookmark_id = :bookmark_id
    AND org_id = :org_id

And finally, we can update our main.py file to pass in these extra fields:

@app.get("/{org_id}/{bookmark_id}")
async def redirect_to_bookmark(
    bookmark_id: str,
    org_id: str,
    current_user: User = Depends(auth.require_user)
):
    # Make sure the user is in the organization
    if user.get_org(org_id) is None:
        raise HTTPException(status_code=401, detail="Not a member of this org")

    link = queries.get_link(org_id=org_id, bookmark_id=bookmark_id)
    if link is None:
        raise HTTPException(status_code=404, detail="Bookmark not found")
    return {"link": link}
    
@app.post("/{org_id}/bookmark")
async def create_bookmark(
    bookmark: Bookmark, 
    org_id: str,
    current_user: User = Depends(auth.require_user)
):
    # Make sure the user is in the organization
    if user.get_org(org_id) is None:
        raise HTTPException(status_code=401, detail="Not a member of this org")

    bookmark_id = generate()
    queries.save_bookmark(bookmark_id=bookmark_id, link=bookmark.link, org_id=org_id, user_id=user.user_id)
    return {"bookmark_id": bookmark_id}

And while this is pretty powerful, we have a bit too much boilerplate if all our routes need explicit authorization checks. In the next section, we’ll add checking roles and refactor our code for readability.

Adding Roles & Permissions (RBAC) as a Dependency

So far, all our routes have been as simple as “if you are in the organization, you can do it.” Let’s look at a route that’s a bit more complicated - deleting bookmarks.

You should be able to delete bookmarks if:

  • You created them
  • You are the Admin of the bookmark’s organization / tenant

A verbose way to make this route is:

@app.delete("/{org_id}/bookmark/{bookmark_id}")
async def delete_bookmark(
    org_id: str, 
    bookmark_id: str, 
    user: User = Depends(auth.require_user)
):
    # First we make sure the user is in that organization
    user_info_in_org = user.get_org(org_id)
    if user_info_in_org is None:
        raise HTTPException(status_code=401, detail="Not a member of this org")

    affected_rows = 0
    if user_info_in_org.is_at_least_role("Admin"):
        # Admins can delete no matter what
        affected_rows = queries.delete_bookmark(
            org_id=org_id, bookmark_id=bookmark_id
        )
    else:
        # Everyone else can only delete their own bookmarks
        affected_rows = queries.delete_my_bookmark(
            org_id=org_id, 
            user_id=user.user_id, 
            bookmark_id=bookmark_id
        )
    
    if affected_rows > 0:
        return {"status": "deleted"}
    else:
        raise HTTPException(status_code=404, detail="Bookmark not found")

Where delete_bookmark and delete_my_bookmark are almost identical, but one takes in a user_id and the other doesn’t.

-- :name delete_bookmark :affected
DELETE FROM bookmarks WHERE bookmark_id = :bookmark_id AND org_id = :org_id;

-- :name delete_my_bookmark :affected
DELETE FROM bookmarks WHERE bookmark_id = :bookmark_id AND org_id = :org_id AND user_id = :user_id;

Since we are using the same authorization checks at the top of every route, a nice way to clean this up would be to use a Dependency.

Here’s what that looks like:

def require_org_access(minimum_required_role: str = None):
    # This is a dependency factory
    # It returns another function that FastAPI calls
    def _require_org_access(org_id: str, user: User = Depends(auth.require_user)):
        user_info_in_org = user.get_org(org_id)
        if user_info_in_org is None:
            raise HTTPException(status_code=401, detail="Not a member of this org")
        
        if minimum_required_role is not None and \\
            not user_info_in_org.is_at_least_role(minumum_required_role):
            raise HTTPException(status_code=403, detail="Insufficient role in org")

        return (user, org)
        
    return _require_org_access

And then we can hook this up in our routes, like this:

@app.post("/{org_id}/bookmark")
async def create_bookmark(
    bookmark: Bookmark, 
    org_id: str,
    user_and_org = Depends(require_org_access())
):
    user, org = user_and_org
    bookmark_id = generate()
    queries.save_bookmark(bookmark_id=bookmark_id, link=bookmark.link, org_id=org_id, user_id=user.user_id)
    return {"bookmark_id": bookmark_id}

Wrapping up

We’ve now built a production-ready, multi-tenant “bookmark aggregator” using FastAPI, dbmate, PugSQL, PropelAuth, and nanoid. In just a few steps, you can handle database migrations, secure your routes by organization membership and roles, and keep your SQL clean and maintainable. FastAPI’s built-in support for Pydantic means you’ll never wrestle with request validation or data integrity concerns again.

From here, the sky’s the limit. Maybe add tagging for bookmarks, or store user analytics. The best part is how easily you can extend all of this. Happy coding!

Sign up for more like this.

Enter your email
Subscribe

Ready to get started?

Sign Up