Production-ready backend with dbmate, PugSQL, PropelAuth, and FastAPI
Tags: webdev, python, beginners, tooling
Published: 2022-06-22 22:33:39

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:

![Our finished API](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/x0wt5ihudre2wo8d0z49.png)

### 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](https://github.com/amacneil/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:

```bash
$ 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.

```sql
-- 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:

```bash
DATABASE_URL="sqlite:db/database.sqlite3"
```

And then we can run our migration with:

```bash
$ 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](https://fastapi.tiangolo.com/) are honestly really good, so we’ll skip repeating those steps. After running `pip install`, let’s start with a basic application:

```python
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 http://localhost:8000/someslug, we need to look up **someslug** and redirect them to the correct URL.

For this, we’ll use [PugSQL](https://pugsql.org/). 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:

```bash
$ pip install pugsql
```

And then update our main.py

```python
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`:

```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`

```sql
-- :name save_url :insert
INSERT INTO urls (slug, url) VALUES (:slug, :url);
```

And then create our save route in **main.py**:

```python
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](https://sive.rs/rand1).

### Testing with OpenAPI

We should have everything we need to test it. Run the server and navigate to [http://localhost:8000/docs](http://localhost:8000/docs)

![First draft of our api](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ozvi1cbziva12o7syxhp.png)

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:

```json
{"slug": "iKzD"}
```

Then navigate to [http://localhost:8000/iKzD](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](https://docs.propelauth.com/getting-started/backend-integration/fastapi/) for FastAPI and start with a pip install:

```bash
$ pip install propelauth_fastapi 
```

Next we initialize the library at the top of our **main.py**

```python
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: 

1. Verify the request is made from a valid user
2. Take in an organization id in the request and verify the user is a member of that organization

```python
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](https://docs.propelauth.com/reference/backend-apis/fastapi/)** 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](https://docs.propelauth.com/overview/access-token/).

**[auth.require_org_member](https://docs.propelauth.com/reference/backend-apis/fastapi/)** 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](http://localhost:8000/docs), you’ll see a new button now in the top right:

![API with authorize button](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/uxmux9oqv6ngn65affvh.png)

The authorize button allows us to provide user tokens so we can continue to test. See [these docs](https://docs.propelauth.com/getting-started/backend-integration/fastapi/#usage-with-api-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:

```bash
$ dbmate new add_org_id
```

And then in that file, we add our migration:

```sql
-- 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:

```bash
$ 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:

```sql
-- :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:

```python
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.

```python
@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`:

```sql
-- :name list_urls :many
SELECT slug, url FROM urls WHERE org_id = :org_id;
```

and `delete_url.sql`

```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.