FastAPI 用户注册与登录实战：JWT 身份认证与权限控制

在前面的教程中，我们已经学习了如何使用 FastAPI 创建 Web API，以及如何连接 MySQL 实现增删改查（CRUD）操作。

但在实际项目中，仅仅能够操作数据库还不够。用户系统通常还需要解决以下问题：

• 用户如何注册账号？

• 如何安全地存储用户密码？

• 用户登录成功后，如何保持身份认证状态？

• 如何限制未登录用户访问指定接口？

• 如何区分普通用户和管理员？

• JWT 令牌过期后应该如何处理？

本文将使用 FastAPI + MySQL + SQLAlchemy 2.0 + JWT，从零搭建一个完整的用户认证示例，并实现注册、登录、获取当前用户信息和管理员权限验证。

一、项目技术栈

本教程使用以下技术：

JWT 的全称是 JSON Web Token。它是一种用于传递声明信息的令牌格式，可以让客户端在后续请求中携带令牌，服务器验证令牌后识别用户身份。

需要注意：JWT 本身不等于加密，也不代表用户已经通过服务器验证。 本文使用签名令牌，并在每次受保护的请求中验证令牌。

二、创建项目并安装依赖

首先创建项目目录：

mkdir fastapi-auth
cd fastapi-auth

建议使用 Python 虚拟环境隔离依赖。

Linux 或 macOS：

python3 -m venv .venv
source .venv/bin/activate

Windows：

python -m venv .venv
.venv\Scripts\Activate.ps1

安装依赖：

pip install fastapi "uvicorn[standard]" sqlalchemy pymysql PyJWT "pwdlib[argon2]" python-multipart

其中：

• sqlalchemy：数据库 ORM。

• pymysql：连接 MySQL。

• PyJWT：处理 JWT 令牌。

• pwdlib[argon2]：对密码进行安全哈希。

• python-multipart：解析登录表单数据。

三、创建 MySQL 数据库

登录 MySQL：

CREATE DATABASE fastapi_auth
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;

创建专门用于应用程序的数据库账号，不建议直接使用 MySQL 的 root 账号运行 Web 应用。

示例：

CREATE USER 'fastapi_user'@'localhost'
IDENTIFIED BY '替换为强数据库密码';

GRANT SELECT, INSERT, UPDATE, DELETE
ON fastapi_auth.*
TO 'fastapi_user'@'localhost';

如果应用和数据库不在同一台服务器上，需要根据实际网络环境调整账号允许连接的来源，并限制数据库端口的访问范围。

四、配置环境变量

为了避免数据库密码和 JWT 密钥直接写进源代码，我们使用环境变量保存敏感配置。

Linux 或 macOS：

export DATABASE_URL='mysql+pymysql://fastapi_user:替换为强数据库密码@127.0.0.1:3306/fastapi_auth?charset=utf8mb4'
export JWT_SECRET_KEY="$(python -c 'import secrets; print(secrets.token_hex(32))')"

Windows PowerShell：

$env:DATABASE_URL = "mysql+pymysql://fastapi_user:替换为强数据库密码@127.0.0.1:3306/fastapi_auth?charset=utf8mb4"
$env:JWT_SECRET_KEY = python -c "import secrets; print(secrets.token_hex(32))"

以上命令用于当前终端会话。实际部署时，应使用服务器的环境变量或安全的密钥管理方案，并确保密钥在应用重启后保持不变。

不要在生产环境中每次启动应用时重新生成 JWT 密钥。 否则，之前签发的令牌会全部失效。

五、创建数据库连接模块

在项目根目录创建 database.py：

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    pool_recycle=1800,
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

这里主要定义了三个对象：

• engine：管理数据库连接。

• SessionLocal：创建数据库会话。

• get_db()：通过 FastAPI 依赖注入，为每次请求提供数据库会话，并在请求结束后关闭会话。

六、设计用户数据表

创建 models.py：

from sqlalchemy import Boolean, Integer, String
from sqlalchemy.orm import Mapped, mapped_column

from database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        index=True,
    )

    username: Mapped[str] = mapped_column(
        String(50),
        unique=True,
        index=True,
        nullable=False,
    )

    email: Mapped[str] = mapped_column(
        String(255),
        unique=True,
        index=True,
        nullable=False,
    )

    hashed_password: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
    )

    is_active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        nullable=False,
    )

    is_admin: Mapped[bool] = mapped_column(
        Boolean,
        default=False,
        nullable=False,
    )

这里有几个需要注意的设计点。

第一，用户名和邮箱需要唯一。 这样可以避免多个账号使用同一个用户名或邮箱。数据库唯一约束才是并发情况下保证唯一性的最终保障，接口层的预先查询不能完全代替它。

第二，只保存 hashed_password。 不要创建明文密码字段，也不要将用户提交的原始密码写入日志。

第三，普通用户不能自行指定管理员身份。 is_admin 应由可信的后台管理流程控制，而不是接受注册请求传入的值。

七、创建请求和响应模型

创建 schemas.py：

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserRegister(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)


class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    is_active: bool
    is_admin: bool

    model_config = ConfigDict(from_attributes=True)


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"

由于 EmailStr 需要额外的邮箱验证依赖，因此再安装：

pip install email-validator

请求模型用于校验客户端提交的数据，响应模型则控制返回给客户端的字段。

尤其需要注意：绝不能将密码哈希直接返回给客户端。

八、实现密码哈希和 JWT 工具

创建 security.py：

import os
from datetime import datetime, timedelta, timezone

import jwt
from pwdlib import PasswordHash

JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

password_hash = PasswordHash.recommended()


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(
    plain_password: str,
    hashed_password: str,
) -> bool:
    return password_hash.verify(
        plain_password,
        hashed_password,
    )


def create_access_token(user_id: int) -> str:
    now = datetime.now(timezone.utc)

    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + timedelta(
            minutes=ACCESS_TOKEN_EXPIRE_MINUTES
        ),
    }

    return jwt.encode(
        payload,
        JWT_SECRET_KEY,
        algorithm=JWT_ALGORITHM,
    )


def decode_access_token(token: str) -> dict:
    return jwt.decode(
        token,
        JWT_SECRET_KEY,
        algorithms=[JWT_ALGORITHM],
        options={"require": ["sub", "exp"]},
    )

为什么要对密码进行哈希？

密码哈希与普通加密不同。密码哈希用于保存密码的验证结果，通常不提供将哈希值还原为原始密码的功能。

当用户登录时，程序通过 verify_password() 比较用户提交的密码和数据库中的哈希值，而不是直接比较明文密码。

Argon2 是专门用于密码哈希的算法之一。相比自行设计加密方式，使用经过验证的密码哈希库更可靠。

JWT 中保存哪些信息？

本文仅在 JWT 中保存：

• sub：用户 ID。

• iat：令牌签发时间。

• exp：令牌过期时间。

不要把密码、密码哈希或其他敏感个人信息放入 JWT。令牌的签名可以防止未经授权的修改，但其载荷通常可以被解码读取。

九、实现用户注册接口

创建 main.py：

from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy import or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from database import Base, engine, get_db
from models import User
from schemas import UserRegister, UserResponse
from security import hash_password

app = FastAPI(title="FastAPI 用户认证系统")

# 仅用于本教程的快速演示。
# 正式项目建议使用 Alembic 管理数据库迁移。
Base.metadata.create_all(bind=engine)


@app.post(
    "/auth/register",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
)
def register(
    data: UserRegister,
    db: Session = Depends(get_db),
):
    username = data.username.strip()
    email = str(data.email).strip().lower()

    if not username:
        raise HTTPException(
            status_code=400,
            detail="用户名不能为空",
        )

    existing_user = db.scalar(
        select(User).where(
            or_(
                User.username == username,
                User.email == email,
            )
        )
    )

    if existing_user:
        raise HTTPException(
            status_code=409,
            detail="用户名或邮箱已被使用",
        )

    user = User(
        username=username,
        email=email,
        hashed_password=hash_password(data.password),
    )

    db.add(user)

    try:
        db.commit()
        db.refresh(user)
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=409,
            detail="用户名或邮箱已被使用",
        )

    return user

注册接口的处理流程如下：

1. 校验用户名、邮箱和密码格式。

2. 检查用户名或邮箱是否已被使用。

3. 对密码进行哈希。

4. 创建用户并写入数据库。

5. 返回用户的公开信息。

这里同时捕获了 IntegrityError，因为两个并发请求可能同时通过预先查询，最终仍需要由数据库唯一约束保证数据一致性。

在面向公众的生产系统中，还应根据隐私和业务要求考虑是否统一错误信息，避免通过注册接口探测某个邮箱是否已经注册，并增加必要的限流和滥用防护。

十、实现用户登录接口

在 main.py 中继续添加：

from fastapi.security import OAuth2PasswordRequestForm

from schemas import TokenResponse
from security import create_access_token, verify_password


@app.post("/auth/login", response_model=TokenResponse)
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db),
):
    user = db.scalar(
        select(User).where(
            User.username == form_data.username.strip()
        )
    )

    if (
        user is None
        or not verify_password(
            form_data.password,
            user.hashed_password if user else "",
        )
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="账号已停用",
        )

    access_token = create_access_token(user.id)

    return TokenResponse(access_token=access_token)

登录接口使用 OAuth2PasswordRequestForm 接收表单数据，因此客户端需要提交：

• username：用户名。

• password：密码。

这里返回的令牌有效期为 30 分钟。生产项目可以根据安全要求调整时长。

另外，OAuth2 密码模式主要适用于理解简单的第一方示例流程；如果开发面向第三方应用或公开客户端的正式 OAuth 登录系统，应优先评估授权码模式与 PKCE 等更合适的方案。

十一、通过依赖注入获取当前登录用户

生成 JWT 只是第一步。后续请求还需要验证令牌，并确认对应用户仍然存在且处于可用状态。

在 main.py 中继续添加：

import jwt

from fastapi.security import OAuth2PasswordBearer
from security import decode_access_token

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/auth/login"
)


def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    credentials_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="登录凭证无效或已过期",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = decode_access_token(token)
        subject = payload.get("sub")

        if not isinstance(subject, str) or not subject.isdigit():
            raise credentials_error

        user_id = int(subject)

    except (jwt.InvalidTokenError, ValueError):
        raise credentials_error

    user = db.get(User, user_id)

    if user is None or not user.is_active:
        raise credentials_error

    return user

OAuth2PasswordBearer 会从请求头中提取 Bearer 令牌。get_current_user() 再完成以下操作：

1. 验证 JWT 的签名和有效期。

2. 读取 sub 中的用户 ID。

3. 从数据库查询用户。

4. 检查用户是否存在、是否已停用。

5. 返回当前用户对象。

因为每次受保护的请求都会重新查询用户状态，所以管理员停用账号后，该账号不能仅凭旧令牌继续访问受保护接口。

十二、创建需要登录才能访问的接口

现在创建一个只允许登录用户访问的接口：

@app.get("/users/me", response_model=UserResponse)
def read_current_user(
    current_user: User = Depends(get_current_user),
):
    return current_user

未登录的请求无法访问该接口。登录成功后，需要在请求头中携带令牌：

Authorization: Bearer 你的访问令牌

如果令牌有效，接口将返回当前用户的公开信息。

FastAPI 的依赖注入机制使得我们不必在每个接口中重复编写 JWT 验证逻辑。

十三、实现管理员权限控制

身份认证解决的是“你是谁”，权限控制解决的是“你能做什么”。

在 main.py 中添加管理员依赖：

def get_current_admin(
    current_user: User = Depends(get_current_user),
) -> User:
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="权限不足",
        )

    return current_user

然后创建一个管理员专用接口：

@app.get("/admin/dashboard")
def admin_dashboard(
    current_admin: User = Depends(get_current_admin),
):
    return {
        "message": "欢迎进入管理后台",
        "admin_id": current_admin.id,
        "username": current_admin.username,
    }

访问该接口需要同时满足两个条件：

1. 用户已经登录，并持有有效令牌。

2. 用户在数据库中的 is_admin 值为 True。

普通用户即使能够自行构造 HTTP 请求，也不能通过修改请求参数将自己变成管理员，因为权限判断读取的是数据库中的用户信息，而不是客户端提交的角色。

如何设置第一个管理员？

本教程的注册接口不会接受 is_admin 参数，因此普通用户不能通过注册请求赋予自己管理员权限。

实际部署时，可以通过受控的数据库初始化脚本或后台管理命令设置首个管理员。例如，在确认对应账号后，由具有权限的维护人员执行：

UPDATE users
SET is_admin = TRUE
WHERE username = '你的管理员用户名';

执行前应确认用户名对应的是正确账号。不要将此类管理员提升功能暴露为任何人都能调用的公开接口。

十四、运行项目并测试接口

启动服务：

uvicorn main:app --reload

浏览器访问：

http://127.0.0.1:8000/docs

FastAPI 会自动生成 Swagger API 文档，方便测试注册、登录和受保护接口。

1. 注册账号

调用：

POST /auth/register
Content-Type: application/json

请求体：

{
  "username": "testuser",
  "email": "test@example.com",
  "password": "ReplaceWithAStrongPassword123"
}

成功后，接口返回用户 ID、用户名、邮箱及账号状态，不会返回密码或密码哈希。

2. 登录获取 JWT

调用 POST /auth/login，提交表单格式的数据：

username=testuser
password=ReplaceWithAStrongPassword123

登录成功后，响应类似：

{
  "access_token": "服务器生成的JWT令牌",
  "token_type": "bearer"
}

实际返回的 access_token 是一串较长的签名令牌，上面仅展示响应结构。

3. 获取当前用户

调用：

GET /users/me
Authorization: Bearer 你的访问令牌

令牌有效时，接口返回当前登录用户的信息。

4. 测试管理员接口

调用：

GET /admin/dashboard
Authorization: Bearer 你的访问令牌

普通用户会收到 403 Forbidden。只有已经被授予管理员权限的账号才能访问成功。

十五、常见错误及解决方法

1. 登录时报 422 错误

常见原因是请求格式不正确。

/auth/login 使用 OAuth2PasswordRequestForm，需要提交表单数据，而不是直接发送 JSON 对象。

2. 报错提示找不到 email-validator

安装依赖：

pip install email-validator

3. 报错提示找不到 python-multipart

安装：

pip install python-multipart

该依赖用于解析登录表单。

4. JWT 令牌始终验证失败

检查以下项目：

• 签发令牌与验证令牌使用的密钥是否相同。

• JWT 算法是否一致。

• 令牌是否已过期。

• 请求头是否采用 Authorization: Bearer ... 格式。

• 服务器的系统时间是否准确。

5. 修改代码后数据库表没有更新

Base.metadata.create_all() 主要用于创建不存在的表，不会自动完成所有表结构迁移。

开发和生产项目建议使用 Alembic 管理数据库结构变更。

6. 数据库出现重复用户名或邮箱

确保数据库中存在唯一约束。接口预先查询只能改善用户体验，不能代替数据库的并发一致性保障。

十六、生产环境还需要哪些安全措施？

本文实现的是一个便于学习的基础认证系统。如果要将其部署到真实网站，还需要进一步完善。

1. 全站使用 HTTPS

不要通过明文 HTTP 传输密码和访问令牌。正式部署时应使用有效的 TLS 证书，并将 HTTP 请求安全地重定向到 HTTPS。

2. 保护 JWT 密钥

使用强随机密钥，通过环境变量或安全的密钥管理服务加载。不要将密钥提交到 Git 仓库，也不要写入公开的配置文件。

3. 增加登录限流

对登录和注册接口设置合理的频率限制，防范密码猜测、撞库和自动化注册。

4. 根据需要实现刷新令牌

访问令牌不宜无限期有效。如果需要长时间保持登录状态，可以设计独立的刷新令牌机制，并考虑撤销、轮换、过期时间和设备管理。

5. 明确令牌存储策略

浏览器应用需要根据架构评估令牌存储方式。将令牌保存在可被 JavaScript 读取的位置会增加 XSS 窃取风险；使用 Cookie 时，则需要妥善配置 HttpOnly、Secure、SameSite 及相应的 CSRF 防护。

6. 完善密码策略

根据业务要求限制密码长度，避免超长输入带来的资源消耗，并在安全需求明确时增加密码重置、邮箱验证和异常登录处理。

7. 使用数据库迁移工具

正式项目应使用 Alembic 管理表结构变更，而不是依赖 create_all() 自动建表。

8. 进一步细化权限

如果项目涉及作者、编辑、审核员和管理员，可以从简单的 is_admin 字段逐步扩展为角色权限模型。对于不同资源，还需要检查用户是否拥有对应资源的访问权，不能只验证是否登录。

十七、总结

通过本教程，我们完成了一个 FastAPI 用户认证系统的核心功能：

• 使用 MySQL 和 SQLAlchemy 保存用户信息。

• 使用 Argon2 对用户密码进行哈希处理。

• 使用 PyJWT 签发和验证访问令牌。

• 通过依赖注入获取当前登录用户。

• 保护需要登录才能访问的接口。

• 使用管理员标记实现基础权限控制。

• 通过统一的响应模型避免泄露密码哈希。

对于个人网站、内容管理系统和后台管理平台来说，这些功能是用户体系的重要基础。