fastapi_users_sqlalchemy_user_model_and_database_adapter.py

1import contextlib
2from typing import AsyncGenerator
3
4from fastapi import Depends
5from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
6from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
7from sqlalchemy.orm import DeclarativeBase
8
9DATABASE_URL = "sqlite+aiosqlite:///./test.db"
10
11class Base(DeclarativeBase):
12    pass
13
14class User(SQLAlchemyBaseUserTableUUID, Base):
15    pass
16
17engine = create_async_engine(DATABASE_URL)
18async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
19
20async def create_db_and_tables():
21    async with engine.begin() as conn:
22        await conn.run_sync(Base.metadata.create_all)
23
24async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
25    async with async_session_maker() as session:
26        yield session
27
28async def get_user_db(session: AsyncSession = Depends(get_async_session)):
29    yield SQLAlchemyUserDatabase(session, User)