FastAPI 的即用型和可定制用户管理

开箱即用的 FastAPI 用户管理。该项目可以快速将用户注册和身份认证功能集成到 FastAPI 项目中，支持注册、登录、密码重置、邮箱验证、OAuth2、自定义身份验证等功能。

特征

可扩展的基本用户模型
即用型注册、登录、重置密码和验证电子邮件路由
即用型社交 OAuth2 登录流程
用于在路由中注入当前用户的依赖项可调用项
可插入密码验证
可定制的数据库后端
多个可定制的身份验证后端
完整的 OpenAPI 模式支持，即使有多个身份验证后端

用户模型和数据库

借助带有 asyncio 的 SQLAlchemy ORM，FastAPI User 提供了使用 SQL 数据库所需的工具。

异步驱动程序

要使用 DBMS 需要安装相应的 asyncio 驱动程序。常见的选择是：

对于 PostgreSQL：pip install asyncpg
对于 SQLite：pip install aiosqlite

创建用户模型

对于任何 SQLAlchemy ORM 模型，将创建一个 User 模型。

from typing import AsyncGenerator

from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

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


class Base(DeclarativeBase):
    pass


class User(SQLAlchemyBaseUserTableUUID, Base):
    pass


engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, 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)

实现创建表的函数

现在将创建一个实用函数来创建所有定义的表。

from typing import AsyncGenerator

from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

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


class Base(DeclarativeBase):
    pass


class User(SQLAlchemyBaseUserTableUUID, Base):
    pass


engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, 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)

例如，可以在 FastAPI 应用程序初始化期间调用此函数。

模式

FastAPI 大量使用 Pydantic 模型来验证请求负载并序列化响应。FastAPI User 也不例外，在读取、创建和更新用户时提供代表用户的 Pydantic 模式。

FastAPI User 提供了一个基本结构来满足其需求。它的结构如下：

id( ID) – 用户的唯一标识符。它与你的 ID 类型匹配，例如 UUID 或整数。
email( str) – 用户的电子邮件。验证通过 email-validator.
is_active( bool) – 用户是否处于活动状态。否则，登录和忘记密码请求将被拒绝。默认为 True.
is_verified( bool) – 用户是否经过验证。可选但对 verify 路由器逻辑很有帮助。默认为 False.
is_superuser( bool) – 用户是否是超级用户。对于实现管理逻辑很有用。默认为 False.

定义你的模式

有四种 Pydantic 模型变体作为 mixin 提供：

BaseUser，提供基本字段和验证；
BaseCreateUseremail，专门用于用户注册，由必填项和字段组成 password；
BaseUpdateUser，专门用于用户个人资料更新，增加了一个可选 password 字段；

应该定义每个变体，并从每个 mixin 继承：

import uuid

from fastapi_users import schemas


class UserRead(schemas.BaseUser[uuid.UUID]):
    pass


class UserCreate(schemas.BaseUserCreate):
    pass


class UserUpdate(schemas.BaseUserUpdate):
    pass

获取当前用户示例

获取当前用户（活跃或非活跃）

current_user = fastapi_users.current_user()

@app.get("/protected-route")
def protected_route(user: User = Depends(current_user)):
    return f"Hello, {user.email}"

获取当前活跃用户

current_active_user = fastapi_users.current_user(active=True)

@app.get("/protected-route")
def protected_route(user: User = Depends(current_active_user)):
    return f"Hello, {user.email}"

获取当前活跃且已验证的用户

current_active_verified_user = fastapi_users.current_user(active=True, verified=True)

@app.get("/protected-route")
def protected_route(user: User = Depends(current_active_verified_user)):
    return f"Hello, {user.email}"

获取当前活动的超级用户

current_superuser = fastapi_users.current_user(active=True, superuser=True)

@app.get("/protected-route")
def protected_route(user: User = Depends(current_superuser)):
    return f"Hello, {user.email}"