Skip to content

Database

The most natural way for storing tokens is of course the very same database you're using for your application. In this strategy, we set up a table (or collection) for storing those tokens with the associated user id. On each request, we try to retrive this token from the database to get the corresponding user id.

Configuration

The configuration of this strategy is a bit more complex than the others as it requires you to configure models and a database adapter, exactly like we did for users.

Model

You should define an AccessToken Pydantic model inheriting from BaseAccessToken.

from fastapi_users.authentication.strategy.db import BaseAccessToken


class AccessToken(BaseAccessToken):
        pass

It is structured like this:

  • token (str) – Unique identifier of the token. It's generated automatically upon login by the strategy.
  • user_id (UUID4) – User id. of the user associated to this token.
  • created_at (datetime) – Date and time of creation of the token. It's used to determine if the token is expired or not.

Database adapter

from typing import AsyncGenerator

from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
from fastapi_users_db_sqlalchemy.access_token import (
    SQLAlchemyAccessTokenDatabase,
    SQLAlchemyBaseAccessTokenTable,
)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker

from .models import AccessToken, UserDB

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
Base: DeclarativeMeta = declarative_base()


class UserTable(Base, SQLAlchemyBaseUserTable):
    pass


class AccessTokenTable(SQLAlchemyBaseAccessTokenTable, Base):
    pass


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


async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


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_access_token_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyAccessTokenDatabase(AccessToken, session, AccessTokenTable)

With Tortoise ORM, you need to define a proper Tortoise model for AccessToken and manually specify the user foreign key. Besides, you need to modify the Pydantic model a bit so that it works well with this Tortoise model.

from fastapi_users import models
from fastapi_users.authentication.strategy.db.models import BaseAccessToken
from fastapi_users.db import TortoiseBaseUserModel
from fastapi_users_db_tortoise.access_token import TortoiseBaseAccessTokenModel
from tortoise import fields
from tortoise.contrib.pydantic import PydanticModel


class User(models.BaseUser):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(models.BaseUserUpdate):
    pass


class UserModel(TortoiseBaseUserModel):
    pass


class UserDB(User, models.BaseUserDB, PydanticModel):
    class Config:
        orm_mode = True
        orig_model = UserModel


class AccessTokenModel(TortoiseBaseAccessTokenModel):
    user = fields.ForeignKeyField("models.UserModel", related_name="access_tokens")


class AccessToken(BaseAccessToken, PydanticModel):
    class Config:
        orm_mode = True
        orig_model = AccessTokenModel

from fastapi_users.db import TortoiseUserDatabase
from fastapi_users_db_tortoise.access_token import TortoiseAccessTokenDatabase

from .models import AccessToken, AccessTokenModel, UserDB, UserModel

DATABASE_URL = "sqlite://./test.db"


async def get_user_db():
    yield TortoiseUserDatabase(UserDB, UserModel)


async def get_access_token_db():
    yield TortoiseAccessTokenDatabase(AccessToken, AccessTokenModel)

import motor.motor_asyncio
from fastapi_users.db import MongoDBUserDatabase
from fastapi_users_db_mongodb.access_token import MongoDBAccessTokenDatabase

from .models import AccessToken, UserDB

DATABASE_URL = "mongodb://localhost:27017"
client = motor.motor_asyncio.AsyncIOMotorClient(
    DATABASE_URL, uuidRepresentation="standard"
)
db = client["database_name"]
users_collection = db["users"]
access_tokens_collection = db["access_tokens"]


async def get_user_db():
    yield MongoDBUserDatabase(UserDB, users_collection)


async def get_access_token_db():
    yield MongoDBAccessTokenDatabase(AccessToken, access_tokens_collection)

Strategy

from fastapi import Depends
from fastapi_users.authentication.strategy.db import AccessTokenDatabase, DatabaseStrategy

from .models import AccessToken, UserCreate, UserDB


def get_database_strategy(
    access_token_db: AccessTokenDatabase[AccessToken] = Depends(get_access_token_db),
) -> DatabaseStrategy[UserCreate, UserDB, AccessToken]:
    return DatabaseStrategy(access_token_db, lifetime_seconds=3600)

As you can see, instantiation is quite simple. It accepts the following arguments:

  • database (AccessTokenDatabase): A database adapter instance for AccessToken table, like we defined above.
  • lifetime_seconds (int): The lifetime of the token in seconds.

Why it's inside a function?

To allow strategies to be instantiated dynamically with other dependencies, they have to be provided as a callable to the authentication backend.

As you can see here, this pattern allows us to dynamically inject a connection to the database.

Logout

On logout, this strategy will delete the token from the database.

Back to top