FastAPI学习

常见项目结构

fastapi-project/
├── app
│   ├── main.py                 # 项目全局入口
│   ├── core                    # 核心配置模块
│   │   ├── config.py           # 全局配置、环境变量读取
│   │   └── security.py         # JWT令牌、密码加密
│   ├── api                     # 路由控制层
│   │   └── v1
│   │       └── user.py         # 用户相关接口
│   ├── schemas                 # Pydantic数据校验、序列化层
│   ├── crud                    # 业务逻辑层
│   ├── models                  # ORM数据库模型层
│   ├── db                      # 数据库连接层
│   └── utils                   # 公共工具类
├── .env                        # 环境变量配置文件
└── requirements.txt             # 项目依赖包

main.py 项目入口层

• 初始化 FastAPI 应用实例
• 配置全局跨域中间件
• 统一注册所有业务路由
• 程序启动唯一入口
• 自动创建数据库数据表
• 管理项目全局配置

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.user import router as user_router
from app.db.database import engine, Base

# 程序启动时自动根据ORM模型创建数据表
Base.metadata.create_all(bind=engine)

# 创建FastAPI主应用
app = FastAPI(
    title="FastAPI分层后端项目",
    version="1.0.0"
)

# 配置跨域中间件，解决前端跨域请求问题
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"]
)

# 注册所有业务路由
app.include_router(user_router)

# 根路径测试接口
@app.get("/")
def index():
    return {"msg":"项目启动成功"}

api 路由控制层 Controller

• 接收前端所有 HTTP 请求
• 匹配请求路径、请求方式
• 获取请求参数、请求头 Token
• 做登录鉴权依赖注入
• 调用业务层方法，不写业务逻辑、不操作数据库
• 接收业务结果，返回统一数据

from fastapi import APIRouter, Depends, Header, HTTPException
from sqlalchemy.orm import Session
from jose import jwt, JWTError

from app.db.database import get_db
from app.schemas.user import UserCreate, TokenResponse
from app.crud.user import create_user, get_user_by_username, verify_login
from app.core.security import create_access_token
from app.utils.response import success, fail
from app.core.config import settings

# 创建用户模块路由对象
router = APIRouter(
    prefix="/user",
    tags=["用户管理模块"]
)

# 用户注册接口
@router.post("/register")
def register(user: UserCreate, db: Session = Depends(get_db)):
    # 判断用户名是否已经存在
    exist_user = get_user_by_username(db, user.username)
    if exist_user:
        return fail(msg="用户名已存在")
    # 调用业务层完成注册
    create_user(db, user)
    return success(msg="注册成功")

# 用户登录接口
@router.post("/login", response_model=TokenResponse)
def login(user: UserCreate, db: Session = Depends(get_db)):
    # 校验账号密码
    login_user = verify_login(db, user.username, user.password)
    if not login_user:
        raise HTTPException(status_code=400, detail="账号或密码错误")
    # 生成登录令牌
    token = create_access_token(login_user.username)
    return success(data={"token": token})

# 需要登录才能访问的个人信息接口
@router.get("/info")
def get_user_info(token: str = Header(None), db: Session = Depends(get_db)):
    try:
        # 解密token获取用户信息
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username = payload.get("sub")
        user = get_user_by_username(db, username)
        return success(data=user)
    except JWTError:
        return fail(msg="Token失效，请重新登录", code=401)

schemas 数据模型层 Pydantic

• 自动校验前端传入参数类型、是否必填
• 自动解析 JSON 请求体
• 格式化、过滤返回给前端的数据
• 配合框架自动生成接口文档
• 提前拦截非法参数，减少业务代码判断
• 实现 ORM 对象与 JSON 数据转换

from pydantic import BaseModel

# 用户注册、登录 请求参数模型
# 自动校验参数类型、必填项
class UserCreate(BaseModel):
    username: str
    password: str

# 用户信息返回模型，规范返回给前端的数据字段
class UserInfo(BaseModel):
    id: int
    username: str

    # 开启ORM映射，支持直接读取数据库对象
    class Config:
        from_attributes = True

# 登录返回token数据模型
class TokenResponse(BaseModel):
    token: str

crud 业务逻辑层 Service

• 编写项目所有业务逻辑

• 数据判断、逻辑处理、业务规则判断

• 调用数据库模型完成增删改查

• 项目真正处理业务的核心代码层

  from sqlalchemy.orm import Session
  from app.models.user import User
  from app.schemas.user import UserCreate
  from app.core.security import get_password_hash, verify_password
  
  # 根据用户名查询用户
  def get_user_by_username(db: Session, username: str):
      return db.query(User).filter(User.username == username).first()
  
  # 用户注册业务逻辑
  def create_user(db: Session, user: UserCreate):
      # 密码加密处理
      hash_pwd = get_password_hash(user.password)
      # 创建数据库用户对象
      db_user = User(username=user.username, password=hash_pwd)
      # 添加数据
      db.add(db_user)
      # 提交事务保存到数据库
      db.commit()
      # 刷新对象数据
      db.refresh(db_user)
      return db_user
  
  # 用户登录账号密码校验
  def verify_login(db: Session, username: str, password: str):
      user = get_user_by_username(db, username)
      # 用户不存在直接返回空
      if not user:
          return None
      # 密码不一致返回空
      if not verify_password(password, user.password):
          return None
      return user

models ORM 数据库模型层

• 映射 MySQL 真实数据表
• 定义数据表字段、类型、主键、备注
• 使用对象操作代替原生 SQL 语句
• 防止 SQL 注入，简化数据库操作
• 与数据库一一对应

from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from app.db.database import Base

# 用户数据表ORM模型，对应mysql user表
class User(Base):
    # 数据库真实表名
    __tablename__ = "user"

    # 用户主键id
    id = Column(Integer, primary_key=True, index=True, comment="用户ID")
    # 用户名，唯一不可重复
    username = Column(String(50), unique=True, nullable=False, comment="用户名")
    # 加密后的用户密码
    password = Column(String(255), nullable=False, comment="用户密码")
    # 账号创建时间
    create_time = Column(DateTime, default=datetime.now, comment="创建时间")

db 数据库连接层

• 创建数据库连接引擎
• 管理数据库会话 Session
• 统一创建、关闭数据库连接
• 提供全局数据库依赖，供所有接口使用

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

# 拼接完整mysql数据库连接地址
SQL_URL = (
    f"mysql+pymysql://{settings.DB_USER}:{settings.DB_PASSWORD}"
    f"@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}?charset=utf8mb4"
)

# 创建数据库引擎，建立数据库基础连接
engine = create_engine(
    SQL_URL,
    pool_pre_ping=True  # 自动检测连接有效性，防止断开报错
)

# 创建数据库会话工厂，用来生成数据库操作对象
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# ORM模型基类，所有数据表模型都继承此类
Base = declarative_base()

# 数据库依赖函数：每次接口请求自动获取一次数据库连接
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        # 请求结束自动关闭连接，释放资源
        db.close()

core 核心配置层

• 读取数据库密钥、JWT 密钥、过期时间
• 密码加密、密码校验工具封装
• JWT 令牌生成、解密、身份认证逻辑

配置文件

# 导入环境变量读取工具
from pydantic_settings import BaseSettings

# 全局配置类，自动读取.env配置文件
class Settings(BaseSettings):
    # 数据库配置
    DB_HOST: str
    DB_PORT: int
    DB_USER: str
    DB_PASSWORD: str
    DB_NAME: str

    # JWT令牌配置
    SECRET_KEY: str
    ALGORITHM: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int

    # 指定配置文件路径
    class Config:
        env_file = ".env"

# 全局唯一配置实例
settings = Settings()

JWT

from datetime import datetime, timedelta
from jose import jwt
from passlib.context import CryptContext
from app.core.config import settings

# 创建密码加密上下文，使用bcrypt加密算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 密码加密：明文密码 -> 加密密码
def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

# 密码校验：明文密码 和 加密密码比对
def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

# 生成JWT登录令牌
def create_access_token(subject: str):
    # 设置token过期时间
    expire_time = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    # 封装令牌载荷数据
    payload = {
        "sub": subject,
        "exp": expire_time
    }
    # 加密生成token字符串
    token = jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return token

utils 工具公共层

• 封装统一接口返回格式
• 统一成功、失败返回体
• 通用工具方法、格式化处理

from fastapi.responses import JSONResponse

# 统一成功返回格式
def success(data=None, msg="请求成功", code=200):
    return JSONResponse(content={
        "code": code,
        "msg": msg,
        "data": data
    })

# 统一失败返回格式
def fail(msg="请求失败", code=400):
    return JSONResponse(content={
        "code": code,
        "msg": msg,
        "data": None
    })

接口完整请求流程

前端请求

→ 跨域中间件校验放行

→ main 主程序路由分发

→ api 路由接收请求

→ schemas 参数自动校验

→ Token 登录身份校验

→ 调用 crud 业务逻辑

→ models 操作数据库

→ db 数据库会话交互

→ 数据封装统一返回前端

→ FastAPI 自动生成接口文档


FastAPI 核心基础知识点

路由请求方式：GET、POST、PUT、DELETE

路径参数、查询参数

请求体 Body（JSON 参数）

Pydantic 数据模型、数据校验

请求头 Header、Cookie

响应数据、自定义返回格式

跨域 CORS

JWT 登录认证 Token

依赖注入 Depends

介绍

依赖注入（Dependency Injection，简称DI），简单来说，就是将程序中「多个模块/接口都需要用到的公共逻辑（如登录校验、数据库连接），抽离成独立的“依赖函数”，在需要使用的地方，通过声明依赖，由框架自动执行该函数、并将执行结果传入，无需手动调用和重复编写的一种编程思想。

在FastAPI中，依赖注入主要通过Depends()方法实现，核心目的是解决代码重复、结构混乱的问题，让接口只专注于核心业务，不用关心公共逻辑的实现细节。

通俗类比：就像写作业时，不用自己准备铅笔、本子（重复杂事），有专门的后勤员（依赖函数）提前准备好，你只要说“我需要学习工具”（声明依赖），后勤员就会把东西递到你手上（自动注入），你直接写作业（核心业务）即可。

实现原理

FastAPI的依赖注入核心是「自动解析+自动执行+自动传值」，整体流程分为3步，简单易懂，无需复杂理解：

1. 定义依赖函数：将公共逻辑（如登录校验、数据库连接）单独写成一个函数，函数可接收参数（如请求头的Token），并返回处理结果（如当前登录用户、数据库会话）。
2. 声明依赖：在接口函数的参数中，通过参数名 = Depends(依赖函数)的方式，告诉FastAPI“这个接口需要用到这个依赖”。
3. 自动注入：当有请求访问接口时，FastAPI会自动先执行声明的依赖函数，获取其返回结果，再将结果自动传入接口函数的对应参数中，接口直接使用该结果即可，无需手动调用依赖函数。

补充：依赖可以多层嵌套（如A依赖B，B依赖C），FastAPI会自动按顺序解析、执行所有依赖，最终将最底层的结果传入接口，全程无需手动干预。同时，依赖函数执行完成后，框架会自动处理资源释放（如数据库连接关闭）。

优缺点

优点

1. 减少重复代码（核心优点）：公共逻辑（登录、连库）只写1次，所有需要的接口直接复用，避免重复编写，提升开发效率。
2. 职责分离，结构清晰：依赖函数负责处理杂事（校验、资源管理），接口只负责核心业务（增删改查），代码分层明确，易读易维护。
3. 统一维护，修改便捷：若公共逻辑需要修改（如登录规则、数据库地址变更），只需修改依赖函数1处，所有依赖该函数的接口自动同步更新，减少修改成本和bug。
4. 实现代码解耦：接口无需关心依赖的内部实现，只需声明“需要什么”，依赖和接口互不影响，便于后续扩展和修改。
5. 便于测试和权限控制：可灵活替换依赖（如测试时用模拟依赖），也可通过多层依赖组合（如“登录校验→管理员校验”），实现灵活的权限管控。
6. 自动管理资源：如数据库连接依赖，框架会自动开启连接、请求结束后自动关闭，避免人为忘记释放资源导致的问题。

缺点

1. 学习成本高：新手难以理解“数据自动传入”的逻辑，阅读代码时需要跳转查看依赖函数，增加阅读成本。
2. 排查问题复杂：若依赖多层嵌套（A→B→C），一旦出现报错，报错链路较长，难以快速定位问题所在。
3. 小型项目易过度设计：若项目只有2-3个接口，无需复杂的公共逻辑，强行使用依赖注入会多写多余代码，小题大做。
4. 存在微量性能损耗：框架需要自动解析依赖、执行依赖函数，比手动调用函数多了一层中间流程，虽损耗极小可忽略，但理论上存在性能开销。
5. 需注意依赖生命周期：不规范使用（如全局依赖未释放资源），可能导致数据库连接泄露、全局变量污染等问题。

异步 async /await

数据库操作、ORM

自动接口文档