fastapi-users-db-sqlalchemy

import os from typing import AsyncGenerator from fastapi import Depends, FastAPI from fastapi_users import FastAPIUsers from fastapi_users.authentication import CookieAuthentication, AuthenticationBackend from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase, Mapped, mapped_column from sqlalchemy import String # Define your database URL (using SQLite for this example) DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///./test.db") SECRET = os.environ.get("SECRET", "YOUR_SECRET_KEY") # IMPORTANT: Change in production! # 1. Define your SQLAlchemy Base class Base(DeclarativeBase): pass # 2. Define your User model class User(SQLAlchemyBaseUserTableUUID, Base): # You can add custom fields here first_name: Mapped[str | None] = mapped_column(String(255), nullable=True) last_name: Mapped[str | None] = mapped_column(String(255), nullable=True) # 3. Create SQLAlchemy engine and session maker engine = create_async_engine(DATABASE_URL) async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # 4. Utility function to create database tables async def create_db_and_tables(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # 5. Database session dependency async def get_async_session() -> AsyncGenerator[AsyncSession, None]: async with async_session_maker() as session: yield session # 6. User database adapter dependency async def get_user_db( session: AsyncSession = Depends(get_async_session), ) -> AsyncGenerator[SQLAlchemyUserDatabase, None]: yield SQLAlchemyUserDatabase(session, User) # 7. Authentication Backend (example with Cookie Authentication) cookie_authentication = CookieAuthentication(secret=SECRET, lifetime_seconds=3600) auth_backend = AuthenticationBackend( name="cookie", transport=cookie_authentication, get_user_manager=lambda user_db: None # UserManager is typically defined outside this module ) # Example FastAPI app (UserManager and other FastAPIUsers components needed for full functionality) app = FastAPI() @app.on_event("startup") async def on_startup(): await create_db_and_tables() # This is a minimal quickstart for the db adapter itself. # For a full working FastAPIUsers example, you'd integrate this with FastAPIUsers, UserManager, and routers. # Example of how you'd use get_user_db in a FastAPI route (simplified): # @app.get("/users/me") # async def get_current_user( # user_db: SQLAlchemyUserDatabase = Depends(get_user_db), # ): # # In a real app, you'd get the current user from auth backend # # and then use user_db to fetch more user data if needed. # # This simply demonstrates the user_db dependency is available. # return {"message": "User database dependency available!"} # To run this: # 1. Save as main.py # 2. pip install fastapi uvicorn 'fastapi-users[sqlalchemy]' aiosqlite # 3. uvicorn main:app --reload