Skip to content
You're not viewing the latest version. Click here to go to latest.

# 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

SQLAlchemyTortoise ORMMongoDB

    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.

model.pyadapter.py

    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