本文档专为前端开发者编写,假设你熟悉 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 项目的每一步,所有代码均可直接复制使用。希望对你有所帮助!

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐