#244: Integrate FastAPI Users Into the To-Do Application

A few weeks ago, we explored two approaches to add authentication to our FastAPI applications. In this post we go a different way and integrate a package that does the heavy lifting for us. I decided to go with FastAPI Users, even when it takes a big step to integrate it.

Installation

We can install FastAPI Users for SQLAlchemy and OAuth with this command:

pip install 'fastapi-users[sqlalchemy,oauth]'

Add the user table

FastAPI Users comes with everything we need, and one of the most obvious things is a table for our users. We can put these few lines to our data/entities.py file to get a basic user table:

from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTableUUID
from sqlalchemy.orm import mapped_column, Mapped
...

class User(SQLAlchemyBaseUserTableUUID, EntityBase):
    pass

If you want to add more fields, you can do that here. For the first round I suggest you start with the built-in parts and only add as soon as you have a working solution.

Create a migration file

With our new table in the entities.py file, we can create a migration file for Alembic with this command:

alembic revision --autogenerate -m "Add User"

The generated migration file should look like this:

"""Add User

Revision ID: 2e5056dec8ae
Revises: 6254e63b6309
Create Date: 2024-09-07 16:04:06.473342

"""
from typing import Sequence, Union

from alembic import op
import fastapi_users_db_sqlalchemy
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '2e5056dec8ae'
down_revision: Union[str, None] = '6254e63b6309'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('user',
    sa.Column('id', fastapi_users_db_sqlalchemy.generics.GUID(), nullable=False),
    sa.Column('email', sa.String(length=320), nullable=False),
    sa.Column('hashed_password', sa.String(length=1024), nullable=False),
    sa.Column('is_active', sa.Boolean(), nullable=False),
    sa.Column('is_superuser', sa.Boolean(), nullable=False),
    sa.Column('is_verified', sa.Boolean(), nullable=False),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_user_email'), table_name='user')
    op.drop_table('user')
    # ### end Alembic commands ###

If all looks good and you added the missing imports (should that be necessary), we can apply the migration with this command:

alembic upgrade head

Do not forget to copy the migrated database to our test template:

cp .\db\todo_api.sqlite .\db\template_test.sqlite

Refactor the dependencies.py file

We need to use the same database file for the FastAPI Users configuration as we do for our data store. But as it stands, we only have a method to get the datastore. We can fix this with a small refactoring:

def db_file():
    db_file = os.path.join(
        os.path.dirname(__file__),
        '.',
        'db',
        'todo_api.sqlite')
    print(f"DB file is: {db_file}")
    return db_file


async def get_db(db_file = Depends(db_file)):
    """
    Creates the datastore 
    """

    factory = await create_async_session_factory(db_file)
    db = DataStoreDb(factory)

    yield db

We can run all our existing tests and they should pass.

Add an authentication.py file

I put all the authentication related code into the authentication.py file. There is a lot going on to glue this plug-in into our application. The main points you should know about are these:

The UserManager class contains all the logic to access the database, works with our User table and uses a GUID as the identifier.
The methods get_async_session() and get_user_db() glue our existing infrastructure together to finally use the get_user_manager() method to initialise the UserManager class.
AuthenticationBackend specifies what we want to use for our authentication (we use JWT with bearer token).
The fastapi_users variable is our entry to the infrastructure (like routers) thar are part of the FastAPI Users package.
We can inject current_active_user into our methods when we want the user to be authenticated.
We put the SECRET_KEY_ENV in a .env file as we did in this post. Make sure that you put that file one directory above the extended_todo folder!

The code to do all that looks like this:

import os
from typing import Optional
import uuid

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin
from fastapi_users.authentication import (
    AuthenticationBackend,
    BearerTransport,
    JWTStrategy,
)
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession

from .data.database import create_async_session_factory
from .data.entities import User
from .dependencies import db_file

from dotenv import load_dotenv
load_dotenv()

SECRET = os.getenv('SECRET_KEY_ENV')


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        print(f"User {user.id} has forgot their password. Reset token: {token}")

    async def on_after_request_verify(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


async def get_async_session(db_file = Depends(db_file)):
    factory = await create_async_session_factory(db_file)
    async with factory() as session:
        yield session


async def get_user_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyUserDatabase(session, User)


async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)


bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)


auth_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)


fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])


current_active_user = fastapi_users.current_user(active=True)

Prepare the endpoint tests

So far, we added all the new parts, but we did not integrate them in our application. Before we put everything together, we need to change our tests. If we are not careful, we can mess up our tests and create race conditions between our test files. To prevent that, we need to create a new fixture that we use for our existing endpoint tests where we assume we have a logged-in user:

test_user = User(email="[email protected]", hashed_password="aaa")

@pytest.fixture(scope="class")
def test_client():
    async def db_file_override():
        db_file = os.path.join(
            os.path.dirname(__file__),
            '..',
            'db',
            'test_db.sqlite')
        return db_file
    app.dependency_overrides[db_file] = db_file_override
    app.dependency_overrides[current_active_user] = lambda: test_user
    client = TestClient(app)
    yield client

As the next step, we must use this fixture in all tests inside test_todo.py and switch from client to test_client:

@pytest.mark.asyncio
async def test_create_task(test_client):
    ...
    response = test_client.post("/api/todo/", json=data)

@pytest.mark.asyncio
async def prepare_task(client, name, priority=4, due_date=None, done=False):
    ...

@pytest.mark.asyncio
async def test_show_task(test_client):
    ...
    id = await prepare_task(test_client, name)

    response = test_client.get(f"/api/todo/{id}")
    ...

# and so on

We can run all our tests and they should pass. We already mock a method we not yet use, but that will change with this new test:

@pytest.mark.asyncio
async def test_abount_me(test_client):
    response = test_client.get("/about/me")
    assert response.status_code == 200
    assert test_user.email in response.text

This new test fails because we do not yet have the endpoint it tries to call. Let us fix that.

Wire-up FastAPI Users in our application

We can now add the 5 routers from FastAPI Users into our application and create a /about/me endpoint to read the email address from the current user:

from .data.entities import User

from .models.user import UserCreate, UserRead, UserUpdate
from .authentication import auth_backend, current_active_user, fastapi_users

...

app.include_router(
    fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
    fastapi_users.get_register_router(UserRead, UserCreate),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(UserRead),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix="/users",
    tags=["users"],
)


@app.get("/about/me")
async def about_me(user: User = Depends(current_active_user)):
    return {"message": f"Hello {user.email}!"}

The FastAPI Users endpoint use these Pydantic models to send data around:

import uuid

from fastapi_users import schemas

class UserRead(schemas.BaseUser[uuid.UUID]):
    pass

class UserCreate(schemas.BaseUserCreate):
    pass

class UserUpdate(schemas.BaseUserUpdate):
    pass

This new code was the missing part for our failing test. We can now rerun all tests and they should pass.

Protect an endpoint

The authentication middleware only helps us when we use it in our application. We can now enforce that the user is logged-in before they can create a new task, update an existing one or delete it. All we need to do is to inject the current_active_user method from our authentication.py file into the endpoint:

from ..authentication import current_active_user
from ..data.entities import User

...

@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskInput, 
                      request: Request, 
                      db: DataStoreDb = Depends(get_db),
                      user: User = Depends(current_active_user)) -> TaskOutput:
    result = await db.add(task)
    headers = {"Location": f"{request.base_url}api/todo/{result.id}"}
    return JSONResponse(content=jsonable_encoder(result), 
                        status_code=status.HTTP_201_CREATED,
                        headers=headers)


@router.put("/{id}")
async def update_task(id: int, 
                      task: TaskInput, 
                      db: DataStoreDb = Depends(get_db),
                      user: User = Depends(current_active_user)) -> TaskOutput:
    try:
        result = await db.update(id, task)
        return result
    except ValueError:
        raise HTTPException(status_code=404, detail="Task not found")


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(id: int, 
                      db: DataStoreDb = Depends(get_db),
                      user: User = Depends(current_active_user)) -> None:
    await db.delete(id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

We can rerun our tests and they will pass. That is because our mock simulates a logged-in user. If we start the application and try to add a task without logging in, we get a 401 error. Let us add some tests to make sure that the login works.

Add authentication tests

The whole logic to log in comes from FastAPI Users. So far, we have no tests to check if the behaviour works as we expect. We can fix that with this set of tests:

from datetime import date, timedelta
import logging
import os

from fastapi.testclient import TestClient
import pytest

from ..data.entities import User

from ..dependencies import db_file
from ..main import app

from ..data.database import create_async_session_factory
from ..data.datastore_db import DataStoreDb
from ..authentication import current_active_user, fastapi_users
logging.getLogger("httpx").setLevel(logging.WARNING)


@pytest.fixture(scope="class")
def test_client():
    async def db_file_override():
        db_file = os.path.join(
            os.path.dirname(__file__),
            '..',
            'db',
            'test_db.sqlite')
        return db_file
    app.dependency_overrides[db_file] = db_file_override
    client = TestClient(app)
    yield client


@pytest.mark.asyncio
async def test_create_task_without_authentication_throws_401_error(test_client):
    data = {
        "name": "A first task",
        "priority": 5,
        "due_date": str(date.today() + timedelta(days=1)),
        "done": False
    }

    response = test_client.post("/api/todo/", json=data)
    assert response.status_code == 401


@pytest.mark.asyncio
async def test_about_me_without_authentication_throws_401_error(test_client):
    response = test_client.get("/about/me")
    assert response.status_code == 401


@pytest.mark.asyncio
async def test_user_can_register(test_client):
    data = {
        "email": "[email protected]",
        "password": "P@ssword123$",
        }
    response = test_client.post("/auth/register", json=data)
    assert response.status_code == 201
    assert response.json()["is_active"] == True
    assert response.json()["is_superuser"] == False
    assert response.json()["is_verified"] == False


@pytest.mark.asyncio
async def test_user_can_login_and_sees_email_in_about_me(test_client):
    data = {
        "email": "[email protected]",
        "password": "P@ssword123$",
        }
    response = test_client.post("/auth/register", json=data)
    assert response.status_code == 201
    print(f"==>'{response.json()}")
    id = response.json()["id"]

    login_data = {
        "username": "[email protected]",
        "password": "P@ssword123$"
        }

    response = test_client.post("/auth/jwt/login", data=login_data)
    response.status_code == 200
    jwt = response.json()["access_token"]

    response = test_client.get("/users/me", headers={"Authorization": "Bearer " + jwt})
    response.status_code == 200
    assert response.json()["email"] == "[email protected]" 
    print(response.json())

    response = test_client.get("/about/me", headers={"Authorization": "Bearer " + jwt})
    response.status_code == 200
    assert response.json()["message"] == "Hello [email protected]!" 
    # print(response.json())

In the test_client we use in this file we are not logged in. Therefore, our attempt to add a task will fail. The same should happen when we call the /about/me endpoint without logging in.

The test test_user_can_register() shows us how we can add a new user. But that will not be enough to work with the API. To see what it takes to log in and use the bearer token, we have the test_user_can_login_and_sees_email_in_about_me() test.

We can now run all the tests and they should pass.

Attention: Uvicorn and .env

If you run the application in Uvicorn, you run it from one directory above the extended_todo folder. If you have the .env file with the SECRET_KEY_ENV variable inside the extended_todo folder, you will get an internal server error with the message "Expected a string value". Good luck with debugging that problem.

The solution to that error is to add the .env file in the parent directory of the extended_todo folder. Then everything works as expected.