Python+FastAPI 搭建 RESTful API —— 把数据库变成接口

📌 适合人群：学完前两篇数据库操作，想知道怎么暴露成 API 的同学📌 你将学到：FastAPI 基础、Pydantic 数据校验、把 SQLAlchemy 模型变成 API 接口📌 建议先读：《Python 操作数据库，这一篇就够了》→《SQLAlchemy ORM》

如果你问我现在 Python 写后端 API 用什么框架，我的答案只有一个——

FastAPI。

为什么？

Flask（2010）   → 老牌经典，但很多事要自己动手Django（2005）  → 大而全，但太重了FastAPI（2018） → 快、自动文档、类型校验、异步支持

而且 FastAPI 和 SQLAlchemy 是天生一对——上一篇文章学到的 ORM 模型，直接就能嵌进 FastAPI 里。

这一篇，我们就用 FastAPI 把之前用户管理系统变成真正的 API 接口。

一、FastAPI 到底有多爽？

先看一段最简单的代码，感受一下：

from fastapi import FastAPIapp = FastAPI()@app.get("/")def read_root():    return {"message": "Hello World"}@app.get("/hello/{name}")def say_hello(name: str):    return {"message": f"你好，{name}！"}

运行：

pip install fastapi uvicornuvicorn main:app --reload

然后浏览器打开：

• • http://127.0.0.1:8000/hello/张三 → {"message": "你好，张三！"}
• • http://127.0.0.1:8000/docs → 自动生成的 Swagger 文档页面，可以直接在浏览器里测试 API！

💡 你没看错——Swagger 文档是自动生成的，不需要写一行文档代码。

二、项目结构

我们用一个完整的项目来展示，结构如下：

user_api/├── main.py          # FastAPI 应用入口 + 路由├── database.py      # 数据库连接├── models.py        # SQLAlchemy 模型（上篇学过的）├── schemas.py       # Pydantic 数据模型（请求/响应校验）└── crud.py          # 数据库操作

这样分层，是实际项目的标准做法，清晰、易维护。

三、一步步搭建

3.1 安装依赖

pip install fastapi uvicorn sqlalchemy pydantic

3.2 database.py — 数据库连接

from sqlalchemy import create_enginefrom sqlalchemy.orm import DeclarativeBase, Session, sessionmaker# SQLite 数据库（开发用）SQLALCHEMY_DATABASE_URL = 'sqlite:///./user_api.db'engine = create_engine(    SQLALCHEMY_DATABASE_URL,    connect_args={'check_same_thread': False}  # SQLite 专用)# 创建会话工厂SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 基类，所有模型都要继承它class Base(DeclarativeBase):    pass# 依赖注入 —— FastAPI 每个请求通过它获取数据库会话def get_db():    db = SessionLocal()    try:        yield db    finally:        db.close()

💡 get_db() 是 FastAPI 的依赖注入机制——每个请求自动获取一个数据库会话，请求结束自动关闭，不用手动管理连接。

3.3 models.py — SQLAlchemy 模型

from sqlalchemy import Column, Integer, Stringfrom database import Baseclass User(Base):    __tablename__ = 'users'    id = Column(Integer, primary_key=True, autoincrement=True)    name = Column(String(50), nullable=False)    age = Column(Integer)    city = Column(String(50))

和上篇文章学的一模一样，直接搬过来用。

3.4 schemas.py — Pydantic 数据模型

这是 FastAPI 的精髓——用 Pydantic 定义请求和响应的数据格式。

from pydantic import BaseModelfrom typing import Optional# ---------- 创建用户时，客户端要传这些 ----------class UserCreate(BaseModel):    name: str    age: Optional[int] = None      # 可选字段    city: Optional[str] = None# ---------- 更新用户时 ----------class UserUpdate(BaseModel):    name: Optional[str] = None     # 全部可选，可以只更新部分字段    age: Optional[int] = None    city: Optional[str] = None# ---------- 返回给客户端的数据格式 ----------class UserResponse(BaseModel):    id: int    name: str    age: Optional[int] = None    city: Optional[str] = None    class Config:        from_attributes = True  # 允许从 ORM 模型自动转换

🔥 Pydantic 自动做数据校验——比如客户端传 age="abc"，FastAPI 自动返回 422 错误，不用你写一行校验代码。

3.5 crud.py — 数据库操作

from sqlalchemy.orm import Sessionfrom models import Userfrom schemas import UserCreate, UserUpdatedef get_users(db: Session, skip: int = 0, limit: int = 100):    """获取用户列表（支持分页）"""    return db.query(User).offset(skip).limit(limit).all()def get_user(db: Session, user_id: int):    """根据 ID 获取单个用户"""    return db.query(User).filter(User.id == user_id).first()def create_user(db: Session, user: UserCreate):    """创建用户"""    db_user = User(name=user.name, age=user.age, city=user.city)    db.add(db_user)    db.commit()    db.refresh(db_user)  # 刷新，拿到数据库生成的自增 ID    return db_userdef update_user(db: Session, user_id: int, user: UserUpdate):    """更新用户"""    db_user = db.query(User).filter(User.id == user_id).first()    if not db_user:        return None    # 只更新传了值的字段    update_data = user.model_dump(exclude_unset=True)    for key, value in update_data.items():        setattr(db_user, key, value)    db.commit()    db.refresh(db_user)    return db_userdef delete_user(db: Session, user_id: int):    """删除用户"""    db_user = db.query(User).filter(User.id == user_id).first()    if not db_user:        return False    db.delete(db_user)    db.commit()    return True

这些代码看着眼熟吗？和上篇 SQLAlchemy 学的一模一样，只是包了一层函数。

3.6 main.py — FastAPI 入口 + 路由

这是最终把一切串起来的地方：

from fastapi import FastAPI, Depends, HTTPExceptionfrom sqlalchemy.orm import Sessionfrom typing import Listfrom database import engine, Base, get_dbfrom models import User          # 导入模型确保表被创建from schemas import UserCreate, UserUpdate, UserResponsefrom crud import get_users, get_user, create_user, update_user, delete_user# 创建所有表Base.metadata.create_all(bind=engine)# 创建 FastAPI 应用app = FastAPI(    title="用户管理系统 API",    description="基于 FastAPI + SQLAlchemy 构建的 RESTful 接口",    version="1.0.0",)# ========== API 路由 ==========@app.get("/")def root():    return {"message": "用户管理系统 API 已启动", "docs": "/docs"}@app.get("/users", response_model=List[UserResponse])def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):    """获取用户列表"""    users = get_users(db, skip=skip, limit=limit)    return users@app.get("/users/{user_id}", response_model=UserResponse)def read_user(user_id: int, db: Session = Depends(get_db)):    """根据 ID 获取用户"""    user = get_user(db, user_id=user_id)    if not user:        raise HTTPException(status_code=404, detail="用户不存在")    return user@app.post("/users", response_model=UserResponse, status_code=201)def create_new_user(user: UserCreate, db: Session = Depends(get_db)):    """创建用户"""    return create_user(db=db, user=user)@app.put("/users/{user_id}", response_model=UserResponse)def update_existing_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):    """更新用户信息"""    db_user = update_user(db=db, user_id=user_id, user=user)    if not db_user:        raise HTTPException(status_code=404, detail="用户不存在")    return db_user@app.delete("/users/{user_id}")def delete_existing_user(user_id: int, db: Session = Depends(get_db)):    """删除用户"""    success = delete_user(db=db, user_id=user_id)    if not success:        raise HTTPException(status_code=404, detail="用户不存在")    return {"message": "删除成功"}

四、跑起来试试

启动服务

cd user_apiuvicorn main:app --reload

看到如下输出表示启动成功：

INFO:     Uvicorn running on http://127.0.0.1:8000INFO:     Application startup complete.

测试 API

方法一：浏览器打开 Swagger 文档

👉 http://127.0.0.1:8000/docs

你会看到一个漂亮的交互式 API 文档页面，每个接口都可以直接在浏览器里点 Try it out 测试。

方法二：用 curl 测试

# 创建用户curl -X POST "http://127.0.0.1:8000/users" \  -H "Content-Type: application/json" \  -d '{"name": "张三", "age": 25, "city": "北京"}'# 创建多个用户curl -X POST "http://127.0.0.1:8000/users" \  -H "Content-Type: application/json" \  -d '{"name": "李四", "age": 30, "city": "上海"}'# 获取用户列表curl "http://127.0.0.1:8000/users"# 获取单个用户curl "http://127.0.0.1:8000/users/1"# 更新用户curl -X PUT "http://127.0.0.1:8000/users/1" \  -H "Content-Type: application/json" \  -d '{"age": 26, "city": "上海"}'# 删除用户curl -X DELETE "http://127.0.0.1:8000/users/2"

方法三：用 Python requests 调用

import requestsBASE = "http://127.0.0.1:8000"# 新增r = requests.post(f"{BASE}/users", json={"name": "赵六", "age": 28, "city": "广州"})print("新增:", r.json())# 查询列表r = requests.get(f"{BASE}/users")print("列表:", r.json())# 查询单个r = requests.get(f"{BASE}/users/1")print("单个:", r.json())# 更新r = requests.put(f"{BASE}/users/1", json={"age": 31})print("更新:", r.json())# 删除r = requests.delete(f"{BASE}/users/3")print("删除:", r.json())

五、再看一次自动文档

FastAPI 最让人上头的特性之一：接口文档自动生成，且完全可交互。

启动服务后打开：

每个接口的：

• • ✅ 请求参数自动校验
• • ✅ 响应格式自动转换成 JSON
• • ✅ 错误状态码自动处理
• • ✅ 接口说明自动生成（从代码里的文档字符串）

这才是真正的"写代码即写文档"。

六、现在，把三篇文章串起来看

这三篇文章其实是一个完整的知识链：

操作数据库（原生 SQL）    →    ORM 操作（SQLAlchemy）    →    API 接口（FastAPI）    │                            │                            │    └── 第1篇                     └── 第2篇                    └── 第3篇（本篇）    sqlite3 / pymysql           User类 / session             /users 接口    cursor.execute()            user.name = "张三"           curl / requests    手动拼接SQL                 操作对象                     自动文档

现在你可以做什么了？

1. 1. 定义好数据库模型（第2篇学的）
2. 2. 写几个 CRUD 函数（第2篇学的）
3. 3. 用 FastAPI 暴露成 API 接口（本篇学的）
4. 4. 前端或者测试工具直接调用

一个真正的后端服务，就这么搭起来了。

七、常见问题

❓ FastAPI 和 Flask 怎么选？

2026 年的今天，新项目我无脑推荐 FastAPI。

❓ 一定要用 SQLite 吗？

当然不。只需改 database.py 里的连接字符串：

# MySQLengine = create_engine('mysql+pymysql://root:123456@localhost/testdb')# PostgreSQLengine = create_engine('postgresql+psycopg2://postgres:123456@localhost/testdb')

其他代码一个字都不用改。

❓ 怎么处理跨域（CORS）？

from fastapi.middleware.cors import CORSMiddlewareapp.add_middleware(    CORSMiddleware,    allow_origins=["*"],           # 允许的前端地址    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],)

加这几行，前端就能跨域调你的 API 了。

八、完整项目代码（直接可运行）

把所有文件汇总成一个 ZIP 的思维，但其实就 5 个文件：

user_api/

├── main.py          # 入口├── database.py      # 连接├── models.py        # 模型├── schemas.py       # 校验└── crud.py          # 操作

复制上面的代码到对应文件，然后：

cd user_apipip install fastapi uvicorn sqlalchemy pydanticuvicorn main:app --reload

打开 http://127.0.0.1:8000/docs，你的第一个 RESTful API 就跑起来了。🎉

总结

三篇连起来看，你从"会写 Python"进阶到了"能搭后端服务"。

📝 整个系列到此完结 🎉

第一篇：Python 操作 5 种数据库（SQLite / MySQL / PostgreSQL / MongoDB / Redis）第二篇：SQLAlchemy ORM —— 不用写 SQL 也能操作数据库第三篇（本篇）：FastAPI —— 把数据库变成 RESTful API

关注我，更多 Python 实战干货持续更新 🚀