对比了 FastAPI、Django 和 Flask 三种 Python Web API 框架组合，分析了各自特点与适用场景。重点阐述了 FastAPI + SQLAlchemy 在异步支持、性能及自动文档方面的优势。教程详细演示了从环境准备、项目结构搭建、MySQL 数据库配置、ORM 模型定义到 Pydantic 数据校验的全过程。内容包括同步 CRUD 操作、基于 BackgroundTasks 的异步批量任务处理、分页查询及 Swagger 文档查看，并提供了 Alembic 迁移建议，适用于构建现代高性能微服务。
FastAPI + SQLAlchemy、Django + Django ORM、Flask + SQLAlchemy是 Python 社区中构建 Web API 的主流选择，各有其适用场景。
async/await）。| 特性 | FastAPI + SQLAlchemy | Django + ORM | Flask + SQLAlchemy |
|---|---|---|---|
| 性能 | ⭐⭐⭐⭐☆ | ⭐⭐ | ⭐⭐⭐ |
| 异步支持 | ✅ 原生支持 | ⚠️ 有限支持 | ❌ |
| 自动文档 | ✅ Swagger/ReDoc | ⚠️ 需第三方 | ⚠️ 需第三方 |
| 学习曲线 | ⚠️ 中等 | ⭐⭐（全栈） | ⭐（简单） |
| 企业级 | ✅ 适合微服务 | ✅ 适合后台系统 | ⚠️ 适合中小型项目 |
结论：如果你要构建现代、高性能、异步支持的 RESTful API 服务，FastAPI + SQLAlchemy 是当前最热门、最具前景的组合之一。
python >=3.8
pip install fastapi[all] sqlalchemy aiomysql pydantic alembic
fastapi_sqlalchemy_tutorial/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── models.py
│ ├── schemas.py
│ ├── database.py
│ └── crud.py
├── alembic/
└── requirements.txt
生成项目结构的 PowerShell 命令（Windows 终端直接复制运行）：
mkdir fastapi_sqlalchemy_tutorial
cd fastapi_sqlalchemy_tutorial
mkdir app
mkdir alembic
New-Item-Path app\__init__.py -ItemType File
New-Item-Path app\main.py -ItemType File
New-Item-Path app\models.py -ItemType File
New-Item-Path app\schemas.py -ItemType File
New-Item-Path app\database.py -ItemType File
New-Item-Path app\crud.py -ItemType File
New-Item-Path requirements.txt -ItemType File
如果不想用 Alembic 自动迁移，可以手动执行以下 SQL 创建表：
CREATE DATABASE IF NOT EXISTS testdb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE testdb;
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
email VARCHAR(100) NOT NULL UNIQUE,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "mysql+aiomysql://user:password@localhost:3306/testdb"
engine = create_engine(DATABASE_URL, echo=True, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
models.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
# ========== ① 用户基础模型 ==========
class UserCreate(BaseModel):
name: str
email: str
class UserUpdate(UserCreate):
id: Optional[int] = None
class UserOut(UserCreate):
id: int
created_at: datetime
class Config:
from_attributes = True
# ========== ② 批量输入模型 ==========
class BatchInOrUpdate(BaseModel):
users: List[UserUpdate]
# ========== ③ 任务输出模型 ==========
class TaskOut(BaseModel):
task_id: str
message: str
# ========== ④ 任务状态模型 ==========
class TaskStatus(BaseModel):
task_id: str
status: str
total: int
from sqlalchemy.orm import Session
from models import User
from database import SessionLocal
from app.schemas import UserCreate, UserUpdate, BatchInOrUpdate
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ================= ① 同步 CRUD =================
def create_user(db: Session, user: UserCreate):
db_user = User(name=user.name, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_users(db: Session, skip: int = 0, limit: int = 10):
return db.query(User).offset(skip).limit(limit).all()
def get_user(db: Session, user_id: int):
return db.query(User).filter(User.id == user_id).first()
def delete_user(db: Session, user_id: int):
user = db.query(User).filter(User.id == user_id).first()
if user:
db.delete(user)
db.commit()
return user
# ================= ② 同步批量插入（后台调用）=================
def batch_insert_update_users(db: Session, users: [UserUpdate]) -> :
u users:
u.:
db.query(User).(User. == u.).update({: u.name, : u.email})
u.:
db.add(User(name=u.name, email=u.email))
db.commit()
(users)
from fastapi import FastAPI, Depends, BackgroundTasks, HTTPException
from sqlalchemy.orm import Session
from app.models import Base, User
from app.database import engine
from app.crud import create_user, get_users, get_user, delete_user, get_db, batch_insert_update_users
from app.schemas import UserCreate, UserOut, BatchInOrUpdate, TaskOut, TaskStatus
import uuid
# 同步建表（demo 阶段够用）
Base.metadata.create_all(bind=engine)
app = FastAPI(title="FastAPI-SQLAlchemy-Tutorial")
# ================= 同步 CRUD =================
@app.post("/users", response_model=UserOut, summary="① 同步 - 创建用户")
def create_user_api(user: UserCreate, db: Session = Depends(get_db)):
return create_user(db, user)
@app.get("/users", response_model=list[UserOut], summary="① 同步 - 列表用户")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
return get_users(db, skip, limit)
@app.get("/users/{user_id}", response_model=UserOut, summary="① 同步 - 查询用户")
def read_user(user_id: int, db: Session = Depends()):
user = get_user(db, user_id)
user:
HTTPException(, )
user
():
user = delete_user(db, user_id)
{: user }
background_tasks = {}
():
background_tasks[] =
total = batch_insert_update_users(db, users)
background_tasks[] =
background_tasks[] = total
():
task_id = (uuid.uuid4())
background_tasks.add_task(run_batch, db, users.users)
{: task_id, : }
():
status = background_tasks.get(, )
total = background_tasks.get(, )
{: task_id, : status, : total}
uvicorn app.main:app --reload
访问：
alembic init alembic
alembic revision --autogenerate -m "init"
alembic upgrade head
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online