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.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
Installation
We can install FastAPI Users for SQLAlchemy and OAuth with this command:
|
1 |
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:
|
1 2 3 4 5 6 |
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:
|
1 |
alembic revision --autogenerate -m "Add User" |
The generated migration file should look like this:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
"""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:
|
1 |
alembic upgrade head |
Do not forget to copy the migrated database to our test template:
|
1 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
@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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
@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:
|
1 2 3 4 5 |
@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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
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 = { "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 = { "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 = { "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 print(response.json()) response = test_client.get("/about/me", headers={"Authorization": "Bearer " + jwt}) response.status_code == 200 # 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.
Next
Integrating FastAPI Users in an application is not a quick task. As this post shows, there are many parts we need to put together before everything works. But as soon as this is done, we can add OAuth logins like Google or Okta to our application without much additional effort.
Next week we put our little to-do application into a Docker container and make it ready for production.
3 thoughts on “Python Friday #244: Integrate FastAPI Users Into the To-Do Application”