Getting Started with FastAPI-Users and Alembic

(Updated 2022-03-15)

FastAPI-Users is a user registration and authentication system that makes adding user accounts to your FastAPI project easier and secure-by-default. It comes with support for various ORMs, and contains all the models, dependencies, and routes you need for registration, activation, email verification, and more.

When setting up your database, you can use SQLAlchemy (or your preferred ORM), plus the provided models, to create the necessary tables very quickly–as you can see from the example in the docs, it doesn’t take much to get everything you need.

In an actively developed project, though, your database is likely to go through many changes over time. Alembic is a tool, used alongside SQLAlchemy, that helps manage database migrations.

This article will cover how to get started with FastAPI-Users and Alembic in a Poetry project. I’ll be using a SQLite database in the examples, because it’s readily available in Python, it’s a good database, and it will illustrate one of Alembic’s features.

Start by running poetry new or poetry init to start a new project.

Adding Dependencies

First of all, let’s add our dependencies to our Poetry project. For the sake of this tutorial, I’ll be pinning specific version numbers. You should consider what versions you want your project to be compatible with when adding your dependencies.

$ poetry add fastapi==0.74.0
$ poetry add fastapi-users[sqlalchemy2]==9.2.5
$ poetry add databases[sqlite]==0.5.5
$ poetry add alembic==1.7.7

Creating the FastAPI App

(Update: since I first wrote this, FastAPI-Users has made some fairly significant changes that make it much more flexible, but require a bit more setup. I’ve kept it as one file below, but I highly recommend seeing the full example in the docs, where it’s separated into different files)

In your project’s source code directory, create a file main.py and put the following code in it.

from typing import AsyncGenerator, Optional

import databases
from fastapi import Depends, FastAPI, Request
from fastapi_users import models as user_models
from fastapi_users import db as users_db
from fastapi_users import BaseUserManager, FastAPIUsers
from fastapi_users.authentication import (
    AuthenticationBackend,
    CookieTransport,
    JWTStrategy,
)
from fastapi_users.db import SQLAlchemyUserDatabase
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
SECRET = "SECRET"


class User(user_models.BaseUser):
    name: Optional[str]


class UserCreate(user_models.BaseUserCreate):
    name: str


class UserUpdate(User, user_models.BaseUserUpdate):
    pass


class UserDB(User, user_models.BaseUserDB):
    pass


database = databases.Database(DATABASE_URL)

Base: DeclarativeMeta = declarative_base()


class UserTable(Base, users_db.SQLAlchemyBaseUserTable):
    name = sa.Column(
        sa.String(length=100),
        server_default=sa.sql.expression.literal("No name given"),
        nullable=False,
    )


engine = create_async_engine(DATABASE_URL, connect_args={"check_same_thread": False})

users = UserTable.__table__
user_db = users_db.SQLAlchemyUserDatabase(UserDB, database, users)


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


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


class UserManager(BaseUserManager[UserCreate, UserDB]):
    user_db_model = UserDB
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

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

    async def on_after_forgot_password(
        self, user: UserDB, 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: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


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


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


app = FastAPI()
fastapi_users = FastAPIUsers(
    get_user_manager,
    [auth_backend],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


app.include_router(
    fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(fastapi_users.get_register_router(), 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(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])

This is basically just the example given by FastAPI-Users, condensed into one file, and minus a few things, including the code that creates the database table–we’ll be using Alembic to do that. If you’re not already familiar with this, I recommend going through the configuration docs where each section of the code is explained.

You can start up the server using poetry run uvicorn projectname.main:app and make sure everything is working. To test it, I navigated to the docs in my browser (http://127.0.0.1:8000/docs). This should show all the FastAPI-Users routes and how to use them. They won’t work yet, since the database isn’t created yet. So, it’s time to set that up!

Initializing Alembic

In the top level directory of your project, run this command:

$ poetry run alembic init alembic

This will create some directories and files. Note that the name passed to the init command can be whatever you want: maybe you’re going to be managing two different databases, and you want to name each directory after the database that it will apply to.

For a description of the files and directories inside the new alembic directory, take a look at the tutorial.

Now, we need to edit alembic.ini to tell it to use our SQLite database. Find the line that looks like

sqlalchemy.url = driver://user:pass@localhost/dbname

and replace it with

sqlalchemy.url = sqlite:///./test.db

Now Alembic is all set up and ready to go!

Creating a Migration Script

We can use the command alembic revision to have Alembic create our first migration script. By passing the -m flag, the script can be titled.

$ poetry run alembic revision -m "Create FastAPI-Users user table"

This should create a file named something like {identifier}_create_fastapi_users_user_table.py, which contains the beginnings of a migration script in it. All we have to do is write the actual migration.

One of the columns in the user table is a custom type, so at the top of the file add this import: from fastapi_users_db_sqlalchemy import GUID

Now, it’s time to write the actual migration! This is in the form of upgrade() and downgrade() functions. The downgrade() function isn’t required, but if you don’t write it, you won’t be able to go revert a database to a previous version.

def upgrade():
    op.create_table( # This tells Alembic that, when upgrading, a table needs to be created.
        "user", # The name of the table.
        sa.Column("id", GUID, primary_key=True), # The column "id" uses the custom type imported earlier.
        sa.Column(
            "email", sa.String(length=320), unique=True, index=True, nullable=False
        ),
        sa.Column("hashed_password", sa.String(length=72), nullable=False),
        sa.Column("is_active", sa.Boolean, default=True, nullable=False),
        sa.Column("is_superuser", sa.Boolean, default=False, nullable=False),
        sa.Column("is_verified", sa.Boolean, default=False, nullable=False),
    )


def downgrade():
    op.drop_table("user") # If we need to downgrade the database--in this case, that means restoring the database to being empty.

Running the Migration

To update to the most recent revision, we can use this command:

$ poetry run alembic upgrade head

It’s also possible to pass specific versions to upgrade/downgrade to, or use relative identifiers to, for example, upgrade to the version two versions ahead of the current one.

If the command was successful, your test.db database file should now contain a user table with all of the columns specified in the script.

Register an Account

To test that the database is now set up, let’s try creating an account.

First, run poetry install to make sure that your project is installed and ready to go. Then, go ahead and start up uvicorn again. You can use anything you want to send the data to the API, but I think the easiest way to verify everything is working is through the docs.

Go to http://127.0.0.1:8000/docs#/auth/register_register_auth_register_post and click on Try it out. Default values are supplied, but you can edit them if you want. When you’re ready, click Execute and see what your server responds with; if all is well, it should be a 201 response code.

Congratulations, you’ve now used Alembic to migrate your database from nonexistence to having a user table suitable for use with FastAPI-Users! From here, you can continue making revisions based on the needs of your project. It’s totally possible to do that by repeating this same process: create a new migration script using alembic revision, then fill in the upgrade() function, then run the migration. However, there is an alternative that many people find easier. Let’s explore that.

Using Alembic’s Autorevision

Suppose, now that you have the basic columns required by FastAPI-Users, you want to update your user table to also have a name column.

When Alembic ran our first migration, it also created its own table to track the current schema and which version of the database it is. When we make changes to our models, we can tell Alembic to compare the new model to the current database, and automatically create a revision based on those changes.

It’s not perfect, and it isn’t intended to be: in some cases, you’ll still need to edit the output to reflect your intentions.

To get started, we need to edit env.py in the alembic directory. All you have to do is import the Base variable from main.py, which should look like from yourproject.main import Base and then assign it to the target_metadata variable. There’s already a line in env.py that looks like target_metadata = None; just change that to target_metadata = Base.metadata. For more detailed instructions, the tutorial has you covered.

Updating the Models

Now that Alembic can see our model, it’s time to actually change the model. To add a name attribute, there are a few lines of code we need to add to main.py.

The Pydantic models User and UserCreate need to be updated. These are what FastAPI will use in establishing what data the API will need to be sent, for example, when the user is created.

Under User, replace pass with the attribute name: Optional[str].

Under UserCreate, replace pass with the attribute name: str.

Now, users will be required to send a name when they make an account, and can optionally update their name when making a patch request to /users/me.

That takes care of the FastAPI side, but we still need to add the column to the SQLAlchemy model. In the UserTable class, replace pass with

    name = sa.Column(
        sa.String(length=100),
        server_default=sa.sql.expression.literal("No name given"),
        nullable=False,
    )

So, on the database side, the name column is going to be a string, not nullable, and with a default value of “No name given”. Adding new NOT NULL columns to an existing table is a tricky business, and having a default value may not be the right way to do it, but that’s another post for another time.

The Migration Script

To generate the migration script, run this command:

$ poetry run alembic revision --autogenerate -m "Added name column"

The new script should look like this:

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user', sa.Column('name', sa.String(length=100), server_default=sa.text("'No name given'"), nullable=False))
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('user', 'name')
    # ### end Alembic commands ###

Looks good! After running the migration with poetry run alembic upgrade head, you can run uvicorn again and check to see that the docs are updated and you’re able to add a user with a name successfully.

Batch Operations

(Update: this section was true when I first wrote it. However, as of version 3.35.0, SQLite supports DROP COLUMN. If you run the below downgrade command with the latest versions of SQLite, it will work instead of failing. I’m leaving this section for historic reasons, and because SQLite still doesn’t support other operations, so the information is still useful. You can check the docs for more information about what is supported and why)

Remember how I said that SQLite would help illustrate one of the features of Alembic? Now is that time.

To see why Batch Operations are necessary, try downgrading the latest revision.

$ poetry run alembic downgrade -1

If you got an exception like sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "DROP": syntax error, then you’ve discovered that SQLite has some limitations when it comes to editing tables in certain ways. One of those limitations is that you can’t drop a column.

Instead, you have to create a new table minus the column you want to drop, copy the data to the new table, drop the old table, and finally rename the new table to the old name.

That sounds like a lot of work, but we can use Alembic’s batch operations to do it for us.

Open up the “Added name column” revision and edit the downgrade() function.

downgrade():
    with op.batch_alter_table("user", schema=None) as batch_op:
        batch_op.drop_column("name")

On Sqlite, this migration will go through the necessary procedure to drop the column. On databases that support dropping columns, it will just drop the column without the extra effort.

There is a lot more to batch operations than that, which you can read all about in the docs, but that covers the basic idea.

If you want, you can have Alembic output autogenerated migration scripts as batch operations by editing env.py and passing the argument render_as_batch=True to context.configure() in the run_migrations_online() function.

Conclusion

In this article, we’ve learned about FastAPI-Users, Alembic, and the basics of how to use Alembic to manage your database. At this point, you should have a functional project with a database ready to accept user registrations. You should have the tools needed to start expanding on that database and upgrade it incrementally as you develop your project.

Finally, here are a few links that I think would be a good place to start learning more.
https://www.chesnok.com/daily/2013/07/02/a-practical-guide-to-using-alembic/
https://speakerdeck.com/selenamarie/alembic-and-sqlalchemy-sane-schema-management?slide=45
https://alembic.sqlalchemy.org/en/latest/cookbook.html#building-an-up-to-date-database-from-scratch
https://www.viget.com/articles/required-fields-should-be-marked-not-null/
https://stackoverflow.com/questions/3492947/insert-a-not-null-column-to-an-existing-table
https://www.red-gate.com/hub/product-learning/sql-prompt/problems-with-adding-not-null-columns-or-making-nullable-columns-not-null-ei028