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.
Database adapters¶
An access token will be structured like this in your database:
token
(str
) – Unique identifier of the token. It's generated automatically upon login by the strategy.user_id
(ID
) – 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.
We are providing a base model with those fields for each database we are supporting.
SQLAlchemy¶
We'll expand from the basic SQLAlchemy configuration.
from typing import AsyncGenerator
from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from fastapi_users_db_sqlalchemy.access_token import (
SQLAlchemyAccessTokenDatabase,
SQLAlchemyBaseAccessTokenTableUUID,
)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
Base: DeclarativeMeta = declarative_base()
class User(SQLAlchemyBaseUserTableUUID, Base):
pass
class AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base): # (1)!
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(session, User)
async def get_access_token_db(
session: AsyncSession = Depends(get_async_session),
): # (2)!
yield SQLAlchemyAccessTokenDatabase(session, AccessToken)
-
We define an
AccessToken
ORM model inheriting fromSQLAlchemyBaseAccessTokenTableUUID
. -
We define a dependency to instantiate the
SQLAlchemyAccessTokenDatabase
class. Just like the user database adapter, it expects a fresh SQLAlchemy session and theAccessToken
model class we defined above.
user_id
foreign key is defined as UUID
By default, we use UUID as a primary key ID for your user, so we follow the same convention to define the foreign key pointing to the user.
If you want to use another type, like an auto-incremented integer, you can use SQLAlchemyBaseAccessTokenTable
as base class and define your own user_id
column.
class AccessToken(SQLAlchemyBaseAccessTokenTable[int], Base):
@declared_attr
def user_id(cls):
return Column(Integer, ForeignKey("user.id", ondelete="cascade"), nullable=False)
Notice that SQLAlchemyBaseAccessTokenTable
expects a generic type to define the actual type of ID you use.
Beanie¶
We'll expand from the basic Beanie configuration.
import motor.motor_asyncio
from beanie import PydanticObjectId
from fastapi_users.db import BeanieBaseUser, BeanieUserDatabase
from fastapi_users_db_beanie.access_token import (
BeanieAccessTokenDatabase,
BeanieBaseAccessToken,
)
DATABASE_URL = "mongodb://localhost:27017"
client = motor.motor_asyncio.AsyncIOMotorClient(
DATABASE_URL, uuidRepresentation="standard"
)
db = client["database_name"]
class User(BeanieBaseUser):
pass
class AccessToken(BeanieBaseAccessToken[PydanticObjectId]): # (1)!
pass
async def get_user_db():
yield BeanieUserDatabase(User)
async def get_access_token_db(): # (2)!
yield BeanieAccessTokenDatabase(AccessToken)
-
We define an
AccessToken
ODM model inheriting fromBeanieBaseAccessToken
. Notice that we set a generic type to define the type of theuser_id
reference. By default, it's a standard MongoDB ObjectID. -
We define a dependency to instantiate the
BeanieAccessTokenDatabase
class. Just like the user database adapter, it expects theAccessToken
model class we defined above.
Strategy¶
import uuid
from fastapi import Depends
from fastapi_users.authentication.strategy.db import AccessTokenDatabase, DatabaseStrategy
from .db import AccessToken, User
def get_database_strategy(
access_token_db: AccessTokenDatabase[AccessToken] = Depends(get_access_token_db),
) -> DatabaseStrategy:
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 forAccessToken
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.