FastAPI 学习总结 - 第三部分：高级应用

十、数据库操作（SQLAlchemy）

安装依赖

pip install sqlalchemy pymysql

数据库配置

知识点详解：

create_engine() 是 SQLAlchemy 创建数据库引擎的核心函数，它负责：

  • 建立与数据库的连接
  • 管理连接池
  • 生成 SQL 语句

连接池参数详解：

参数 值 作用
pool_size=5 5 常驻连接数，避免频繁创建/销毁连接的开销
max_overflow=10 10 当 5 个连接不够时，允许临时创建最多 10 个额外连接
pool_timeout=30 30秒 如果 30 秒内无法获取连接，抛出超时异常

SessionLocal 的作用：

  • 每次请求创建新的数据库会话
  • autocommit=False：需要手动 commit，保证数据一致性
  • autoflush=False：需要手动 flush，避免意外的自动提交
# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接URL
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/fastapi_db"

# 创建数据库引擎
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_size=5,           # 连接池大小
    max_overflow=10,      # 最大溢出连接数
    pool_timeout=30       # 连接超时时间
)

# 创建会话工厂
SessionLocal = sessionmaker(
    autocommit=False,     # 不自动提交
    autoflush=False,      # 不自动刷新
    bind=engine           # 绑定引擎
)

# 创建基类
Base = declarative_base()

模型定义

知识点详解：

Column 参数详解：

参数 说明
primary_key=True 主键，自动建立索引，用于唯一标识记录
index=True 为此列创建 B-Tree 索引，加快查询速度
unique=True 唯一约束，不允许重复值
nullable=False 非空约束
autoincrement=True 自动递增，常用于主键

relationship（关系）的本质：

  • relationship() 不会在数据库中创建列
  • 它是 SQLAlchemy 的"魔法"属性，访问时自动执行 JOIN 查询
  • back_populates 和 backref 互为镜像，建立双向关系

一对一 vs 一对多 vs 多对多：

  • 一对多：relationship("Item", back_populates="owner")
  • 一对一：在一对多基础上加 uselist=False
  • 多对多：需要中间表
# app/models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base
from datetime import datetime

class User(Base):
    """用户模型"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    username = Column(String(50), nullable=False, unique=True, index=True)
    email = Column(String(100), nullable=False, unique=True, index=True)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, default=datetime.now)

    # 一对多关系：一个用户可以有多个商品
    items = relationship("Item", back_populates="owner")

class Item(Base):
    """商品模型"""
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    description = Column(String(500))
    price = Column(Integer, nullable=False)
    owner_id = Column(Integer, ForeignKey("users.id"))  # 外键建立关联

    # 反向关系：通过 item.owner 获取用户
    owner = relationship("User", back_populates="items")

创建表

# 创建所有表
Base.metadata.create_all(bind=engine)

CRUD 操作

知识点详解：

会话（Session）生命周期管理：

db.add(user)      → 将对象标记为"待添加"（状态: pending）
db.commit()       → 提交事务，永久保存到数据库（状态: committed）
db.refresh(user)  → 从数据库重新读取最新数据（如自增ID、默认值）
db.delete(user)   → 标记为待删除
db.rollback()     → 回滚未提交的更改

为什么需要 db.refresh()？

  • 自增 ID 是在 commit() 后由数据库生成的
  • refresh() 确保获取到数据库生成的值

查询执行时机：

  • .all()、.first()、.one() 才会真正执行 SQL
  • 只构建查询对象时不执行

创建数据：

def create_user(db, user_data):
    """创建用户"""
    user = User(
        username=user_data.username,
        email=user_data.email,
        password_hash=hash_password(user_data.password)
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

查询数据：

def get_user(db, user_id):
    """根据ID获取用户"""
    return db.query(User).filter(User.id == user_id).first()

def get_user_by_email(db, email):
    """根据邮箱获取用户"""
    return db.query(User).filter(User.email == email).first()

def get_users(db, skip=0, limit=10):
    """获取用户列表（分页）"""
    return db.query(User).offset(skip).limit(limit).all()

更新数据：

def update_user(db, user_id, user_data):
    """更新用户信息"""
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        if user_data.username:
            user.username = user_data.username
        if user_data.email:
            user.email = user_data.email
        db.commit()
        db.refresh(user)
    return user

删除数据：

def delete_user(db, user_id):
    """删除用户"""
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

复杂查询

知识点详解：

and_ / or_ 组合条件：

  • and_()：所有条件同时满足（相当于 SQL 的 AND）
  • or_()：满足任一条件即可（相当于 SQL 的 OR）
  • 可以嵌套使用，实现复杂逻辑

常见查询方法对比：

方法 说明 返回值
.first() 获取第一条 单条记录或 None
.all() 获取所有 列表
.one() 获取唯一一条 单条记录，多条或零条会报错
.scalar() 获取标量值 第一个字段的值
.count() 计数 整数

多条件查询：

from sqlalchemy import and_, or_

# 查询年龄大于18且邮箱包含example的用户
users = db.query(User).filter(
    and_(
        User.age > 18,
        User.email.like("%example%")
    )
).all()

# 查询用户名是admin或邮箱是admin@example.com的用户
users = db.query(User).filter(
    or_(
        User.username == "admin",
        User.email == "admin@example.com"
    )
).all()

排序：

# 按创建时间降序排序（最新在前）
users = db.query(User).order_by(User.created_at.desc()).all()

# 多字段排序：先按时间降序，再按用户名升序
users = db.query(User).order_by(User.created_at.desc(), User.username.asc()).all()

分页：

page = 1
page_size = 10
# 计算偏移量：(page-1) * page_size
users = db.query(User).offset((page - 1) * page_size).limit(page_size).all()
# page=1: offset(0).limit(10)  获取第1-10条
# page=2: offset(10).limit(10) 获取第11-20条

聚合函数：

from sqlalchemy import func

# 统计用户数量
count = db.query(func.count(User.id)).scalar()

# 计算平均年龄
avg_age = db.query(func.avg(User.age)).scalar()

# 分组统计：按性别统计人数
result = db.query(User.gender, func.count(User.id)).group_by(User.gender).all()

十一、安全认证

知识点详解：

JWT（JSON Web Token）认证流程：

用户登录 → 验证密码 → 生成JWT → 返回token → 后续请求携带token → 验证token → 获取用户信息

JWT 的三部分结构：

xxxxx.yyyyy.zzzzz
 ↑       ↑      ↑
Header  Payload Signature
  • Header：存储算法类型（如 HS256）
  • Payload：存储声明（Claims），如 sub（用户标识）、exp（过期时间）
  • Signature：签名，确保 Token 未被篡改

为什么用 sub 作为用户标识？

  • sub（subject）是 JWT 标准声明
  • 存储用户 email 或 user_id，便于后续通过 sub 查询用户

密码哈希的必要性：

  • 不能明文存储密码（数据库泄露 = 密码泄露）
  • bcrypt 是"慢哈希"，故意设计慢，抵抗暴力破解
  • 同一密码每次哈希结果不同（因为有盐）

OAuth2 密码流

安装依赖：

pip install python-jose[cryptography] passlib[bcrypt]

创建安全模块：

# app/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from app.crud import get_user_by_email

# 密钥（实际应用中应该从环境变量获取）
SECRET_KEY = "your-secret-key-keep-it-safe"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password, hashed_password):
    """验证密码"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    """生成密码哈希"""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None):
    """创建访问令牌"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_db)):
    """获取当前用户"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = get_user_by_email(db, email=email)
    if user is None:
        raise credentials_exception
    return user

登录接口：

# app/api/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta
from app.security import create_access_token, verify_password, ACCESS_TOKEN_EXPIRE_MINUTES
from app.crud import get_user_by_email

router = APIRouter(tags=["认证"])

@router.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get_db)):
    """获取访问令牌"""
    user = get_user_by_email(db, email=form_data.username)
    if not user or not verify_password(form_data.password, user.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.email}, expires_delta=access_token_expires
    )

    return {"access_token": access_token, "token_type": "bearer"}

保护接口：

from app.security import get_current_user
from app.models import User

@router.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    """获取当前用户信息"""
    return {"email": current_user.email, "username": current_user.username}

API 密钥认证

知识点详解：

API Key vs JWT：

  • API Key：简单，适合服务端到服务端
  • JWT：支持过期时间、可携带更多信息、适合跨域

auto_error=False 的作用：

  • 默认为 True：请求头缺失时自动抛出 403 错误
  • 设置为 False：请求头缺失时返回 None，由我们自己处理逻辑
from fastapi import Security, HTTPException
from fastapi.security import APIKeyHeader

# 从环境变量获取API密钥
API_KEY = "your-api-key"

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def get_api_key(api_key: str = Security(api_key_header)):
    """验证API密钥"""
    if api_key != API_KEY:
        raise HTTPException(status_code=403, detail="Invalid API Key")
    return api_key

@app.get("/protected")
async def protected_route(api_key: str = Depends(get_api_key)):
    """受保护的接口"""
    return {"message": "This is a protected route"}

十二、中间件

知识点详解：

中间件执行流程：

请求 → 中间件1 → 中间件2 → ... → 路由处理 → 中间件2 → 中间件1 → 响应

中间件 vs 依赖注入：

  • 中间件：全局影响所有请求，可以修改请求/响应
  • 依赖注入：按需使用，只能处理特定逻辑

自定义中间件

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """记录请求处理时间"""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

CORS 中间件

知识点详解：

CORS（跨域资源共享）原理：

  • 浏览器的同源策略限制 Ajax 只能访问同源资源
  • CORS 通过响应头告诉浏览器是否允许跨域请求

参数详解：

参数 说明
allow_origins 允许的来源，["*"] 表示全部
allow_credentials 是否允许携带 Cookie
allow_methods 允许的 HTTP 方法
allow_headers 允许的请求头
expose_headers 暴露给前端的响应头
max_age 预检请求（OPTIONS）的缓存时间 
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],                    # 允许的来源
    allow_credentials=True,                 # 允许携带凭证
    allow_methods=["*"],                    # 允许的HTTP方法
    allow_headers=["*"],                    # 允许的请求头
    expose_headers=["X-Process-Time"],      # 暴露给前端的响应头
    max_age=600,                           # 预检请求缓存时间
)

日志中间件

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """记录请求日志"""
    print(f"请求方法: {request.method}")
    print(f"请求路径: {request.url}")
    print(f"请求头: {dict(request.headers)}")

    response = await call_next(request)

    print(f"响应状态码: {response.status_code}")
    return response

十三、背景任务

知识点详解：

BackgroundTasks vs Celery：

特性 BackgroundTasks Celery
适用场景 轻量级任务（写日志、发邮件） 重量级/分布式任务
执行位置 进程内 独立进程/服务器
复杂度 低 高（需要消息队列）
失败重试 不支持 支持

为什么用 background_tasks.add_task() 而不是直接调用？

  • 直接调用会阻塞，直到函数执行完毕才返回响应
  • add_task() 将任务添加到队列，立即返回响应

基本用法

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
    """写入日志"""
    with open("log.txt", mode="a") as log:
        log.write(f"{message}\n")

def send_email(email: str, message: str):
    """发送邮件（模拟）"""
    print(f"Sending email to {email}: {message}")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    """发送通知"""
    # 添加后台任务
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    background_tasks.add_task(send_email, email, "Hello from FastAPI!")

    return {"message": "Notification sent"}

带依赖的背景任务

知识点详解：

依赖注入与后台任务的冲突：

问题：后台任务在请求结束后执行，但 get_db 依赖在请求结束时关闭连接。

解决方案：需要将 db 对象传递给后台任务，而不是依赖注入的 db。

def get_db():
    """获取数据库连接"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def process_data(db, data_id: int):
    """处理数据"""
    # 执行耗时操作
    item = db.query(Item).filter(Item.id == data_id).first()
    if item:
        item.processed = True
        db.commit()

@app.post("/process/{data_id}")
async def process_item(
    data_id: int,
    background_tasks: BackgroundTasks,
    db=Depends(get_db)
):
    """处理数据（后台任务）"""
    background_tasks.add_task(process_data, db, data_id)
    return {"message": "Processing started"}

十四、静态文件服务

知识点详解：

app.mount() vs app.add_api_route()：

  • mount()：挂载整个 ASGI 应用（如 StaticFiles），处理所有路径
  • add_api_route()：添加单个路由

为什么叫 “mount”？

  • 类似 Linux 挂载点，将某个路径绑定到指定应用
  • /static 下的所有请求都由 StaticFiles 处理

基本配置

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 挂载静态文件目录
app.mount("/static", StaticFiles(directory="static"), name="static")

目录结构：

project/
├── main.py
└── static/
    ├── images/
    │   └── logo.png
    ├── css/
    │   └── style.css
    └── js/
        └── app.js

访问示例：

  • http://localhost:8000/static/images/logo.png
  • http://localhost:8000/static/css/style.css

HTML 模板

知识点详解：

为什么 TemplateResponse 需要 request 对象？

  • Jinja2 模板可能需要生成 URL（如 url_for()）
  • 传入 request 才能使用 url_for() 和 session 等

Jinja2 模板变量：

  • {{ variable }}：输出变量值
  • {% if %} ... {% endif %}：条件判断
  • {% for %} ... {% endfor %}：循环
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/")
async def read_root(request: Request):
    """渲染首页"""
    return templates.TemplateResponse("index.html", {"request": request, "name": "FastAPI"})

模板文件 templates/index.html：

<!DOCTYPE html>
<html>
<head>
    <title>FastAPI</title>
    <link rel="stylesheet" href="/static/css/style.css">
</head>
<body>
    <h1>Hello, {{ name }}!</h1>
    <img src="/static/images/logo.png" alt="Logo">
</body>
</html>

十五、异步支持

知识点详解：

同步 vs 异步的本质区别：

维度 同步 异步
等待 I/O 时 阻塞线程 切换到其他任务
资源占用 每个请求一个线程 单线程处理多请求
适用场景 CPU 密集型 I/O 密集型（API、数据库）

什么时候用 async/await：

  • I/O 操作：数据库查询、HTTP 请求、文件读写
  • 不需要 CPU 一直计算

什么时候用同步（不用 async）：

  • CPU 密集型：图片处理、数据计算
  • 如果用了 async 但没有 await，代码会立即返回，没有意义

异步路径操作

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
async def read_items():
    """异步获取商品列表"""
    # 模拟异步操作
    items = await fetch_items_from_database()
    return items

async def fetch_items_from_database():
    """从数据库异步获取数据"""
    # 实际应用中使用异步数据库驱动
    return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]

异步依赖

知识点详解：

同步 vs 异步依赖注入：

类型 定义方式 使用场景
同步 def get_db() 同步 ORM（如 SQLAlchemy）
异步 async def get_db() 异步 ORM（如 SQLAlchemy+asyncpg） 
async def get_db():
    """异步获取数据库连接"""
    db = AsyncSessionLocal()
    try:
        yield db
    finally:
        await db.close()

@app.get("/users")
async def read_users(db=Depends(get_db)):
    """异步获取用户列表"""
    users = await db.execute(select(User))
    return users.scalars().all()

异步 HTTP 请求

知识点详解：

为什么用 httpx 而不是 requests？

  • requests：同步库，会阻塞
  • httpx：支持异步，完美配合 FastAPI

async with 的作用：

  • 自动管理资源（连接）
  • 进入时创建，退出时关闭
import httpx

@app.get("/fetch-data")
async def fetch_external_data():
    """异步获取外部API数据"""
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

十六、测试

知识点详解：

为什么用 pytest？

  • 简洁的语法：assert 而不是 self.assertEqual()
  • 丰富的 fixture 支持
  • 强大的参数化功能

TestClient 的注意事项：

  • TestClient 是同步的，即使路由是 async def
  • 如果想测试真正的异步，使用 httpx.AsyncClient

测试金字塔：

        /\
       /  \     E2E 测试（少量）
      /----\
     /      \   集成测试（中等）
    /--------\
   /          \  单元测试（大量）
  /____________\

使用 pytest

安装依赖：

pip install pytest httpx

测试文件 tests/test_main.py：

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    """测试根路径"""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_user():
    """测试创建用户"""
    response = client.post(
        "/users",
        json={"username": "testuser", "email": "test@example.com", "password": "123456"}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "testuser"
    assert data["email"] == "test@example.com"
    assert "password" not in data  # 确保密码不在响应中

def test_read_items():
    """测试获取商品列表"""
    response = client.get("/items?skip=0&limit=10")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

运行测试：

pytest tests/ -v

测试认证接口

知识点详解：

Bearer Token 认证流程：

  1. 登录获取 token
  2. 在请求头 Authorization: Bearer <token> 中携带 token
  3. 后端验证 token 的有效性
def test_login():
    """测试登录"""
    # 先创建用户
    client.post(
        "/users",
        json={"username": "admin", "email": "admin@example.com", "password": "secret"}
    )

    # 测试登录
    response = client.post(
        "/token",
        data={"username": "admin@example.com", "password": "secret"}
    )
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

def test_protected_route():
    """测试受保护的接口"""
    # 获取token
    response = client.post(
        "/token",
        data={"username": "admin@example.com", "password": "secret"}
    )
    token = response.json()["access_token"]

    # 访问受保护的接口
    response = client.get(
        "/users/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    assert response.json()["email"] == "admin@example.com"

十七、项目结构

知识点详解：

分层架构的优点：

优点 说明
单一职责 每层只做一件事
可维护性 修改某一层不影响其他层
可测试性 可以单独测试每一层
可扩展性 易于添加新功能

各层职责详解：

模块 职责 关注点
api/ 定义路由和接口逻辑 HTTP 请求/响应
models/ 定义数据库表结构 数据存储
schemas/ 定义请求/响应数据结构 数据验证和转换
crud/ 实现数据库CRUD操作 业务数据操作
security.py 实现认证和授权逻辑 安全
database.py 数据库连接配置 基础设施

推荐结构

project/
├── main.py                 # 应用入口
├── requirements.txt        # 依赖配置
├── .env                   # 环境变量
├── app/
│   ├── __init__.py
│   ├── main.py            # FastAPI实例配置
│   ├── api/               # 路由接口
│   │   ├── __init__.py
│   │   ├── users.py       # 用户相关接口
│   │   ├── items.py       # 商品相关接口
│   │   └── auth.py        # 认证接口
│   ├── models/            # 数据库模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/           # Pydantic模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── crud/              # 数据操作
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── security.py        # 安全相关
│   └── database.py        # 数据库配置
├── tests/                 # 测试文件
│   ├── __init__.py
│   ├── test_users.py
│   └── test_items.py
└── static/                # 静态文件
    ├── css/
    ├── js/
    └── images/

模块职责说明

模块 职责
api/ 定义路由和接口逻辑
models/ 定义数据库表结构（SQLAlchemy模型）
schemas/ 定义请求/响应的数据结构（Pydantic模型）
crud/ 实现数据库CRUD操作
security.py 实现认证和授权逻辑
database.py 数据库连接配置

十八、部署

知识点详解：

Uvicorn vs Gunicorn：

特性 Uvicorn Gunicorn
类型 ASGI 服务器 WSGI 服务器
协议 ASGI（支持异步） WSGI（仅同步）
Worker 管理 手动 自动
推荐场景 开发/简单部署 生产环境

进程数计算公式：

  • CPU 密集型：2 * CPU 核心数 + 1
  • I/O 密集型：4 * CPU 核心数 + 1

使用 Uvicorn

# 开发模式（代码变更自动重载）
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

# 生产模式
uvicorn main:app --host 0.0.0.0 --port 8000

使用 Gunicorn + Uvicorn

知识点详解：

为什么生产环境用 Gunicorn？

  • 自动管理多个 Worker 进程
  • 处理 Worker 崩溃重启
  • 更稳定的进程管理
# 安装依赖
pip install gunicorn

# 启动服务（4个worker进程）
gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

使用 Docker

知识点详解：

Docker 部署的优势：

  • 环境一致性：开发、测试、生产环境相同
  • 隔离性：应用之间互不影响
  • 快速部署：一键启动

镜像选择建议：

  • python:3.11-slim：生产推荐，体积小
  • python:3.11：体积大，包含完整工具

Dockerfile：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

构建和运行：

# 构建镜像
docker build -t fastapi-app .

# 运行容器
docker run -p 8000:8000 fastapi-app

使用 Nginx 反向代理

知识点详解：

为什么用 Nginx？

用户请求 → Nginx (80/443) → FastAPI (8000)
  • SSL/TLS 终结：HTTPS 加密在 Nginx 处理
  • 静态文件服务：Nginx 比 Python 更高效
  • 负载均衡：多 FastAPI 实例分担压力
  • 请求限流：保护后端服务

关键配置解释：

配置 作用
proxy_pass 转发请求到 FastAPI
proxy_set_header Host $host 传递原始 Host 头
proxy_set_header X-Real-IP 传递真实客户端 IP
proxy_set_header X-Forwarded-For 传递代理链 IP
proxy_set_header X-Forwarded-Proto 传递原始协议（HTTPS）

Nginx配置文件 /etc/nginx/sites-available/fastapi：

server {
    listen 80;
    server_name example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

总结

FastAPI 核心优势回顾：

特性 说明
高性能 基于 Starlette，支持 async/await，媲美 Node.js 和 Go
类型安全 全面支持 Python 类型提示，IDE 自动补全
自动文档 Swagger UI 和 ReDoc 自动生成，开箱即用
依赖注入 强大的 DI 系统，代码复用性强
数据验证 Pydantic 自动验证请求/响应，错误信息清晰

适合场景：

  • RESTful API 开发
  • 微服务架构
  • 实时应用（WebSocket）
  • 数据处理后端

学习路径建议：

第一阶段：基础
├── 路由定义（@app.get/post/put/delete）
├── 参数获取（路径、查询、请求体）
└── Pydantic 数据验证

第二阶段：数据库
├── SQLAlchemy 模型定义
├── CRUD 操作
└── 关系映射（一对多、多对多）

第三阶段：安全
├── JWT 认证流程
├── OAuth2 密码流
└── 密码哈希（bcrypt）

第四阶段：高级
├── 异步编程（async/await）
├── 中间件
├── 背景任务
└── 测试（pytest）

第五阶段：部署
├── Uvicorn/Gunicorn
├── Docker 容器化
└── Nginx 反向代理

快速上手建议：

  1. 从官方文档示例开始，边学边练
  2. 使用 uvicorn app.main:app --reload 开启热重载开发
  3. 使用 http://localhost:8000/docs 查看自动生成的 API 文档
  4. 参考优秀的开源 FastAPI 项目学习最佳实践
  5. 阅读 Starlette 文档深入理解底层原理
# fastapi # 数据库 # python # 开发语言 # mysql # sql
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

当 AI 编程从“辅助”走向“治理”：一个前端工程师的 2026 年观察

2026 年，AI 编程正在从代码补全、AI IDE 进入 Agent 工程化阶段。它不再只是帮开发者写几段代码，而是开始参与 issue 处理、测试修复、重构迁移甚至 PR 提交。但真正的问题也随之变化：AI 是否能生成代码，已经不是核心；更关键的是，它生成的代码是否安全、可维护、可审计。本文结合 METR 开发者生产力研究、Agentic Coding 趋势、MCP 安全风险，以及实际开发中遇

avatar AtomGit开源社区

编程开挂指南：构建你的超能力技能树

《代码开挂指南：IT人的超能力培养体系》摘要 该体系构建了从基础到前沿的立体化技能树：核心层涵盖编程语言、算法、设计模式三大支柱；工具链维度要求精通开发工具、版本控制和调试技术；工程化思维强调规范流程与自动化部署。认知层面培养代码直觉与抽象建模能力，通过快捷键、自动化等技巧提升效能。前瞻性拓展包括AI编程、云原生等未来技术，建议通过技术雷达跟踪、开源贡献和写作输出来持续进化。每个技能点均配备效率对

avatar AtomGit开源社区
cover

Cursor 对接国内大模型（DeepSeek 等）完整部署配置教程

avatar AtomGit开源社区