In this post, we’ll build a fully production-ready backend complete with DB migrations, B2B authentication, to show off all the cool features that FastAPI provides.
The application itself is a B2B URL-shortener. Our customers can shorten URLs, view all the URLs submitted by their coworkers, and admins within the organization will be able to delete URLs.
We’ll build this in as an API-first product, starting with this API:
Setting up the database with dbmate
We need some way to create the initial database, and thinking forward a bit, we’ll also need a way to update the schema as we need in order to make changes to the application.
For this, we’ll use dbmate, which makes it easy for us to run database migrations regardless of what our backend looks like. After installing dbmate, we’ll make our first migration:
$ dbmate new create_url_table
This will create a migrations directory and create a new file with a name that looks like: 20220616004439_create_url_table.sql
Migrations consist of two parts: up and down. Up applies the migration and down reverts it.
-- migrate:up
create table urls (
slug text NOT NULL PRIMARY KEY,
url text NOT NULL
);
-- migrate:down
drop table urls;
This is a pretty common pattern. We create the table in the up and then drop it in the down. dbmate looks for a .env file to know where the database is, so let’s create that first:
DATABASE_URL="sqlite:db/database.sqlite3"
And then we can run our migration with:
$ dbmate up
You should now have a file db/database.sqlite3, which is your database.
Setting up FastAPI and connecting to the database with PugSQL
The FastAPI docs on getting started are honestly really good, so we’ll skip repeating those steps. After running pip install, let’s start with a basic application:
from fastapi import FastAPI, HTTPException
from starlette.responses import RedirectResponse
app = FastAPI()
@app.get("/{slug}")
async def redirect_to_full_url(slug: str):
url = # TODO: how do we get the URL that matches the slug?
if url is None:
raise HTTPException(status_code=404)
return RedirectResponse(url)
This is our core function, when a user visits our site with a url, like localhost:8000/someslug, we need to look up someslug and redirect them to the correct URL.
For this, we’ll use PugSQL. I personally don’t like using ORMs, which is why we aren’t directly using SQLAlchemy here, and prefer to write SQL as much as possible. PugSQL provides that ability.
To understand PugSQL, let’s see it in action, starting with installing it:
$ pip install pugsql
And then update our main.py
import pugsql
from fastapi import FastAPI, HTTPException
from starlette.responses import RedirectResponse
app = FastAPI()
queries = pugsql.module('db/queries')
queries.connect("sqlite:///db/database.sqlite3")
@app.get("/{slug}")
async def redirect_to_full_url(slug: str):
url = queries.get_url(slug=slug)
if url is None:
raise HTTPException(status_code=404)
return RedirectResponse(url)
Ok… there’s a lot of magic going on. We pass in a path db/queries, connect our database, and then magically have a get_url function that takes in a slug?
We are missing one important file. The way PugSQL works is you create SQL files that map to functions. In this case, we need to create a file db/queries/get_url.sql:
-- :name get_url :scalar
SELECT url FROM urls WHERE slug = :slug;
The function name comes from :name, the return value is :scalar meaning a single value, or the URL, and the function argument slug comes from :slug in the sql file.
Before we can test it, we need a way to save URLs, so let’s create db/queries/save_url.sql
-- :name save_url :insert
INSERT INTO urls (slug, url) VALUES (:slug, :url);
And then create our save route in main.py:
class Url(BaseModel):
url: str
@app.post("/url")
async def create_shortened_url(url: Url):
slug = save_url_with_unique_slug(url)
return {"slug": slug}
def save_url_with_unique_slug(url: Url):
num_duplicate_slugs = 0
# Keep creating new slugs until we get a unique one
while True:
slug = generate_random_slug()
try:
queries.save_url(slug=slug, url=url.url)
return slug
except IntegrityError:
# IntegrityError catches the case where we tried to
# insert a slug that already exists
num_duplicate_slugs += 1
# If we get this many collisions, something is wrong
if num_duplicate_slugs > 10:
raise HTTPException(status_code=500)
def generate_random_slug():
slug_length = random.randint(4, 7)
return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(slug_length))
Most of the code exists to create a random slug and check if it exists in the DB already. You can also do this directly in the database.
Testing with OpenAPI
We should have everything we need to test it. Run the server and navigate to http://localhost:8000/docs
Our API is live! This is still my favorite FastAPI feature because now we have a full OpenAPI spec generated and live. If you make a post request to the /url endpoint, you’ll get back a response like:
{"slug": "iKzD"}
Then navigate to http://localhost:8000/iKzD and you will be redirected to the original URL you entered before.
Adding authentication and organizations with PropelAuth
Our plan initially was to build a B2B application however, at this stage, we don’t have any concept of users or the companies/organizations they are members of.
PropelAuth is a user auth service that makes it easy to add auth to any B2B application. We’ll follow our getting started guide for FastAPI and start with a pip install:
$ pip install propelauth_fastapi
Next we initialize the library at the top of our main.py
from propelauth_fastapi import init_auth, User
auth = init_auth("YOUR_AUTH_URL", "YOUR_API_KEY")
And finally, we’ll update our create_shortened_url function to do two things:
- Verify the request is made from a valid user
- Take in an organization id in the request and verify the user is a member of that organization
class Url(BaseModel):
url: str
org_id: str
@app.post("/url")
async def create_shortened_url(url: Url, user: User = Depends(auth.require_user)):
auth.require_org_member(user, url.org_id, minimum_required_role=None)
slug = save_url_with_unique_slug(url)
return {"slug": slug}
auth.require_user verifies the request is made from a valid user. For more details about how this works under the hood, check out PropelAuth’s documentation on access tokens.
auth.require_org_member verifies that the user is a member of whichever org_id was specified in the request. We’ll allow anyone in the org to create a url, but if we wanted to limit it to specific roles, we could have by specifying the minimum_required_role.
PropelAuth’s hosted pages allow our users to manage their own accounts and organizations, so you don’t need to worry about building any of that.
Now if you revisit http://localhost:8000/docs, you’ll see a new button now in the top right:
The authorize button allows us to provide user tokens so we can continue to test. See these docs, which cover how to use this button with PropelAuth.
We haven’t, however, updated the database to use our new org_id, so it’s time to write a new migration!
Adding org_id to our schema
Just like before, let’s run:
$ dbmate new add_org_id
And then in that file, we add our migration:
-- migrate:up
ALTER TABLE urls ADD COLUMN org_id TEXT;
-- migrate:down
ALTER TABLE urls DROP COLUMN org_id;
We made the org_id optional for backwards compatibility reasons, but if you prefer you can just delete the data and add NOT NULL. Apply these changes with dbmate:
$ dbmate up
Next, we need to update our save function to take into account org_ids, so let’s revisit the save_url.sql file:
-- :name save_url :insert
insert into urls (slug, url, org_id) values (:slug, :url, :org_id);
And finally, update our save_url call to also pass in the org_id:
queries.save_url(slug=slug, url=url.url, org_id=url.org_id)
Fetching all URLs within an organization
Now, when a user saves a URL to the database, it will also save the user’s organization. But we aren’t really using that org_id yet. Let’s add a few quick routes for fetching all the URLs within an organization and deleting a URL.
@app.get("/org/{org_id}/url")
async def list_urls_for_org(org_id: str, user: User = Depends(auth.require_user)):
auth.require_org_member(user, org_id, minimum_required_role=None)
return queries.list_urls(org_id=org_id)
@app.delete("/org/{org_id}/url/{slug}")
async def delete_url(org_id: str, slug: str, user: User = Depends(auth.require_user)):
auth.require_org_member(user, org_id, minimum_required_role=UserRole.Admin)
queries.delete_url(org_id=org_id, slug=slug)
Seems easy enough. We take the org_id in the path this time, verify only valid users can call it, verify the user is a member of that organization, and then perform their action. We’re just missing the two sql files, list_urls.sql:
-- :name list_urls :many
SELECT slug, url FROM urls WHERE org_id = :org_id;
and delete_url.sql
-- :name delete_url :affected
DELETE FROM urls WHERE slug = :slug AND org_id = :org_id;
Summary
By choosing these libraries and services, we were able to build and test an entire B2B application in record time:
- dbmate allowed us to get our database up and running quickly, and when we needed to modify the schema it was just as easy.
- PugSQL allowed us to write regular SQL queries that looked just like python functions - which provides a really nice separation of concerns.
- PropelAuth allowed us to not worry about any auth details and just focus on if our users are logged in and what their roles within an organization are.
- FastAPI let us test, even with authentication in place, via their build in OpenAPI UIs.
If you want to see how extensible it is, try adding a few extra features. For example, how quickly can you add analytics per URL? If this walkthrough was useful for your company, let us know what other apps you want to see built.