FastAPI 后端项目从零到一开发文档(前端开发者友好版)
可以尝试点击 `POST /users/` 接口,输入 JSON 数据(如 `{"name": "张三", "age": 25}`),测试用户创建功能。或者单独安装 `python-multipart`(用于表单处理)和 `fastapi` 自带的 CORS 中间件,但最简单的是安装 `fastapi[all]`。再次访问 `http://127.0.0.1:8000/docs`,你会看到新增的
本文档专为前端开发者编写,假设你熟悉 JavaScript 但无后端开发经验。我们将使用 Python 的 FastAPI 框架,配合 MySQL 数据库,一步步搭建一个包含用户和文章管理的 RESTful API 服务。每一步都有详细说明,并提供完整代码示例。
第一步:搭建 Python 开发环境并安装 FastAPI
1.1 检查 Python 版本
打开终端(命令提示符或 PowerShell),输入:
python --version
或
python3 --version
需要 Python 3.9 或更高版本。如果没有安装,请前往 [Python 官网](https://www.python.org/downloads/) 下载并安装,安装时务必勾选 **“Add Python to PATH”**。
1.2 创建项目文件夹
在合适的位置创建项目根目录,例如:
mkdir fastapi-study
cd fastapi-study
1.3 创建虚拟环境
虚拟环境可以隔离项目依赖,避免与系统其他 Python 项目冲突。
# Windows
python -m venv venv
# Mac/Linux
python3 -m venv venv
这会在当前文件夹下生成一个 `venv` 子文件夹。
1.4 激活虚拟环境
Windows (cmd 或 PowerShell):
venv\Scripts\activate
Mac/Linux:
source venv/bin/activate
激活后,终端提示符前会出现 `(venv)` 字样。
1.5 安装 FastAPI 和 Uvicorn
Uvicorn 是一个轻量级的 ASGI 服务器,用于运行 FastAPI 应用。
pip install fastapi uvicorn
1.6 编写第一个 FastAPI 应用并测试
在项目根目录创建 `main.py`,写入以下内容:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello World"}
启动服务:
uvicorn main:app --reload
终端会显示类似 `Uvicorn running on http://127.0.0.1:8000` 的信息。打开浏览器访问 `http://127.0.0.1:8000`,如果看到 `{"message":"Hello World"}`,说明环境搭建成功。
第二步:连接 MySQL 数据库并配置 SQLAlchemy
2.1 确保 MySQL 服务已启动并创建数据库
检查 MySQL 是否运行(Windows 可在服务中查看,Mac/Linux 使用 `brew services list` 或 `systemctl status mysql`)。
-登录 MySQL 并创建数据库 `study`(如果尚未创建):
mysql -u root -p
# 输入密码 123456
CREATE DATABASE study;
exit
2.2 安装数据库驱动和 ORM
在虚拟环境中安装 `sqlalchemy` 和 `pymysql`:
pip install sqlalchemy pymysql
2.3 创建数据库配置文件 `database.py`
在项目根目录新建 `database.py`,配置数据库连接:
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接 URL:mysql+pymysql://用户名:密码@主机:端口/数据库名
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/study"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
2.4 测试数据库连接
创建测试文件 `test_db.py`:
运行测试:
python test_db.py
如果看到“数据库连接成功!”,说明配置正确。
第三步:创建数据模型和 CRUD 接口3.1 定义数据模型 `models.py`
创建 `models.py`,定义 `User` 表:
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
age = Column(Integer, nullable=True)
3.2 定义 Pydantic 模型 `schemas.py`
创建 `schemas.py`,用于请求和响应的数据验证:
from pydantic import BaseModel
from typing import Optional
class UserBase(BaseModel):
name: str
age: Optional[int] = None
class UserCreate(UserBase):
pass
class User(UserBase):
id: int
class Config:
3.3 封装数据库操作 `crud.py`
创建 `crud.py`,实现用户的增删改查:
from sqlalchemy.orm import Session
import models
import schemas
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
db_user = models.User(name=user.name, age=user.age)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def update_user(db: Session, user_id: int, user: schemas.UserCreate):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if db_user:
db_user.name = user.name
db_user.age = user.age
db.commit()
db.refresh(db_user)
return db_user
def delete_user(db: Session, user_id: int):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if db_user:
db.delete(db_user)
db.commit()
return db_user
3.4 编写 API 路由 `main.py`
替换 `main.py` 为以下完整内容
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import engine, SessionLocal
import models
import schemas
import crud
# 创建数据库表(如果不存在)
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# 依赖项:获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/")
def read_root():
return {"message": "Hello World"}
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
return crud.create_user(db=db, user=user)
@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
@app.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.update_user(db=db, user_id=user_id, user=user)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
@app.delete("/users/{user_id}", response_model=schemas.User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.delete_user(db=db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
3.5 启动服务并测试
重新启动服务:
uvicorn main:app --reload
打开浏览器访问 `http://127.0.0.1:8000/docs`,你会看到 FastAPI 自动生成的交互式 API 文档(Swagger UI)。可以尝试点击 `POST /users/` 接口,输入 JSON 数据(如 `{"name": "张三", "age": 25}`),测试用户创建功能。如果成功,返回的数据会包含自动生成的 `id`。
第四步:配置 CORS 并添加文章模型
4.1 安装 CORS 支持
为了允许前端应用(如运行在 `http://localhost:3000` 的 React 或 Vue 项目)访问 API,需要配置 CORS。在虚拟环境中安装包含 CORS 的完整依赖:
pip install fastapi[all]
或者单独安装 `python-multipart`(用于表单处理)和 `fastapi` 自带的 CORS 中间件,但最简单的是安装 `fastapi[all]`。
4.2 修改 `main.py` 添加 CORS 中间件
在 `main.py` 顶部导入:
from fastapi.middleware.cors import CORSMiddleware
在创建 `app` 对象后添加:
app = FastAPI()
# CORS 配置
origins = [
"http://localhost:3000", # React/Vue 开发服务器
"http://127.0.0.1:3000",
"http://localhost:8000", # 允许自身
# 可以添加更多前端地址
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 允许的域名列表
allow_credentials=True,
allow_methods=["*"], # 允许所有 HTTP 方法
allow_headers=["*"], # 允许所有请求头
)
4.3 在 `models.py` 中添加文章模型 `Post`
修改 `models.py`,增加文章表并与用户建立一对多关联:
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
age = Column(Integer, nullable=True)
# 关联到文章:一个用户有多篇文章
posts = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(100), nullable=False)
content = Column(String(500))
user_id = Column(Integer, ForeignKey("users.id"))
# 关联到用户
author = relationship("User", back_populates="posts")
4.4 在 `schemas.py` 中添加文章相关的 Pydantic 模型
from pydantic import BaseModel
from typing import Optional, List
# 用户模型(扩展,包含文章列表)
class UserBase(BaseModel):
name: str
age: Optional[int] = None
class UserCreate(UserBase):
pass
class User(UserBase):
id: int
posts: List["Post"] = [] # 嵌套文章列表
class Config:
orm_mode = True
# 文章模型
class PostBase(BaseModel):
title: str
content: Optional[str] = None
class PostCreate(PostBase):
user_id: int
class Post(PostBase):
id: int
user_id: int
class Config:
orm_mode = True
# 解决前向引用(Pydantic v2 可能需要)
User.update_forward_refs()
4.5 在 `crud.py` 中添加文章 CRUD 操作
# 文章相关操作
def get_post(db: Session, post_id: int):
return db.query(models.Post).filter(models.Post.id == post_id).first()
def get_posts(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Post).offset(skip).limit(limit).all()
def create_post(db: Session, post: schemas.PostCreate):
db_post = models.Post(title=post.title, content=post.content, user_id=post.user_id)
db.add(db_post)
db.commit()
db.refresh(db_post)
return db_post
def update_post(db: Session, post_id: int, post: schemas.PostCreate):
db_post = db.query(models.Post).filter(models.Post.id == post_id).first()
if db_post:
db_post.title = post.title
db_post.content = post.content
db_post.user_id = post.user_id
db.commit()
db.refresh(db_post)
return db_post
def delete_post(db: Session, post_id: int):
db_post = db.query(models.Post).filter(models.Post.id == post_id).first()
if db_post:
db.delete(db_post)
db.commit()
return db_post
4.6 在 `main.py` 中添加文章路由
在 `main.py` 的现有路由之后添加:
@app.post("/posts/", response_model=schemas.Post)
def create_post(post: schemas.PostCreate, db: Session = Depends(get_db)):
return crud.create_post(db=db, post=post)
@app.get("/posts/", response_model=list[schemas.Post])
def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
return crud.get_posts(db, skip=skip, limit=limit)
@app.get("/posts/{post_id}", response_model=schemas.Post)
def read_post(post_id: int, db: Session = Depends(get_db)):
db_post = crud.get_post(db, post_id=post_id)
if db_post is None:
raise HTTPException(status_code=404, detail="Post not found")
return db_post
@app.put("/posts/{post_id}", response_model=schemas.Post)
def update_post(post_id: int, post: schemas.PostCreate, db: Session = Depends(get_db)):
db_post = crud.update_post(db=db, post_id=post_id, post=post)
if db_post is None:
raise HTTPException(status_code=404, detail="Post not found")
return db_post
@app.delete("/posts/{post_id}", response_model=schemas.Post)
def delete_post(post_id: int, db: Session = Depends(get_db)):
db_post = crud.delete_post(db=db, post_id=post_id)
if db_post is None:
raise HTTPException(status_code=404, detail="Post not found")
return db_post
4.7 自动创建新表
由于我们修改了模型,重启服务后 `models.Base.metadata.create_all(bind=engine)` 会自动创建 `posts` 表(如果不存在)。无需额外操作。
4.8 重启服务并测试
uvicorn main:app --reload
再次访问 `http://127.0.0.1:8000/docs`,你会看到新增的 `/posts/` 相关接口。测试流程:
1. 先通过 `POST /users/` 创建一个用户,记下返回的 `id`。
2. 通过 `POST /posts/` 创建一篇文章,请求体中包含 `user_id` 字段。
3. 通过 `GET /users/{id}` 查看该用户时,应该能看到其文章列表(`posts` 字段)。
第五步:验证接口并排查常见错误
5.1 在 Swagger UI 中测试
打开 `http://127.0.0.1:8000/docs`。
点击任意接口,再点击 **Try it out**。
输入参数(对于 POST 请求,需要在请求体中输入 JSON),点击 **Execute**。
查看响应状态码和返回数据。
5.2 常见错误及解决方法
-
500 Internal Server Error:查看运行 FastAPI 的终端窗口中的详细错误信息。常见原因包括:
-
数据库连接失败(检查
database.py中的 URL 是否正确,MySQL 服务是否启动)。 -
表未创建(可手动登录 MySQL 执行
SHOW TABLES;检查;若未创建,可手动运行建表语句或确保create_all被调用)。 -
外键约束失败(创建文章时提供的
user_id在users表中不存在)。
-
-
422 Unprocessable Entity:请求体格式错误,例如缺少必填字段、字段类型不对。请检查 JSON 是否合法(键名必须用双引号,字符串值也用双引号,不能有多余逗号)。
-
404 Not Found:查询的资源不存在,属于正常业务逻辑。
-
JSON decode error:请求体不是合法的 JSON,通常是因为语法错误(如使用单引号、缺少逗号等)。请确保 JSON 格式正确。
5.3 手动检查数据库表
如果怀疑表未自动创建,可以登录 MySQL 确认:
mysql -u root -p
# 输入密码
USE study;
SHOW TABLES;
如果表不存在,可以手动执行以下 SQL 创建:
CREATE TABLE users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
age INT
);
CREATE TABLE posts (
id INT AUTO_INCREMENT PRIMARY KEY,
title VARCHAR(100) NOT NULL,
content VARCHAR(500),
user_id INT,
FOREIGN KEY (user_id) REFERENCES users(id)
);
5.4 成功验证
当你能够成功创建用户、创建文章,并且通过列表接口和详情接口都能正确获取数据时,说明你的 FastAPI 后端已经搭建成功,可以供前端调用了。
本文档完整记录了从零开始搭建 FastAPI 项目的每一步,所有代码均可直接复制使用。希望对你有所帮助!
更多推荐
所有评论(0)