FastAPI + SQLAlchemy 构建高性能 RESTful API 接口实战
FastAPI + SQLAlchemy、Django + Django ORM、Flask + SQLAlchemy是 Python 社区中构建 Web API 的主流选择,各有其适用场景。
1. FastAPI + SQLAlchemy
- 特点:异步支持、性能高、自动文档生成(基于 OpenAPI)、类型注解驱动。
- 适用场景:高并发、微服务、现代 Python 项目。
- 优势:
- 原生异步支持(
async/await)。
- 自动生成 Swagger / ReDoc 文档。
- 与 Pydantic 无缝集成,数据校验强。
- 劣势:
- 相对较新,生态不如 Django 成熟。
- 对初学者来说,异步和依赖注入有一定学习曲线。
2. Django + Django ORM
- 特点:全栈框架、ORM 强大、后台管理、认证系统完善。
- 适用场景:企业级后台系统、内容管理系统、快速开发。
- 优势:
- 一体化解决方案,开发效率高。
- 自带后台管理、认证、权限系统。
- ORM 功能强大,迁移系统成熟。
- 劣势:
- 相对重量级,不适合轻量级 API 服务。
- 异步支持较弱(Django 3+ 支持异步视图,但仍不完善)。
3. Flask + SQLAlchemy
- 特点:轻量、灵活、插件丰富。
- 适用场景:中小型项目、原型开发、教学用途。
- 优势:
- 简单易学,社区庞大。
- 可自由组合组件(如 Flask-RESTful、Flask-JWT)。
- 劣势:
- 缺乏现代异步支持。
- 需要手动集成很多功能(如认证、文档、验证等)。
4. 总结对比
| 特性 | FastAPI + SQLAlchemy | Django + ORM | Flask + SQLAlchemy |
|---|
| 性能 | ⭐⭐⭐⭐☆ | ⭐⭐ | ⭐⭐⭐ |
| 异步支持 | ✅ 原生支持 | ⚠️ 有限支持 | ❌ |
| 自动文档 | ✅ Swagger/ReDoc | ⚠️ 需第三方 | ⚠️ 需第三方 |
| 学习曲线 | ⚠️ 中等 | ⭐⭐(全栈) | ⭐(简单) |
| 企业级 | ✅ 适合微服务 | ✅ 适合后台系统 | ⚠️ 适合中小型项目 |
结论:如果你要构建现代、高性能、异步支持的 RESTful API 服务,FastAPI + SQLAlchemy 是当前最热门、最具前景的组合之一。
5. 环境准备
python >=3.8
pip install fastapi[all] sqlalchemy aiomysql pydantic alembic
6. 项目结构
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
7. 数据库配置(MySQL)
如果不想用 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)
8. 定义模型(SQLAlchemy)
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())
9. Pydantic 校验模型(schemas.py)
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
10. CRUD 操作(crud.py)
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()
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)
11. 主程序(main.py)
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
Base.metadata.create_all(bind=engine)
app = FastAPI(title="FastAPI-SQLAlchemy-Tutorial")
@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}
12. 运行与文档查看
uvicorn app.main:app --reload
访问:
13. 使用 Alembic 做迁移(可选)
alembic init alembic
alembic revision --autogenerate -m "init"
alembic upgrade head
14. 总结与扩展建议
- 可添加认证(JWT)
- 可集成 Redis 缓存
- 可部署到 Docker + Uvicorn + Gunicorn