Manuel Ernesto ⚡️
FastAPI 从入门到项目实战：AI 掘金头条完整学习笔记

FastAPI 从入门到项目实战：AI 掘金头条完整学习笔记

2026/6/23
Project Learning FastAPIPythonSQLAlchemyRedisVueBackend

FastAPI 从入门到项目实战：AI 掘金头条完整学习笔记

适合对象：会 Python，但刚开始学习 Web API、FastAPI、前后端分离项目的同学。
这份笔记基于黑马程序员 FastAPI 教程项目「AI 掘金头条」、实际运行排错过程，以及 FastAPI / SQLAlchemy / Pydantic / Uvicorn / Redis / Passlib 等官方资料整理。
推荐学习方式：先跑通最小 FastAPI，再拆项目结构，再学用户模块，最后补 Redis、异常处理、部署与排错。

目录

1. FastAPI 到底解决什么问题

你以前写 Python，大多数代码是本地调用：

def add(a, b):
    return a + b

print(add(1, 2))

这叫本地函数调用。调用方和函数都在同一个 Python 程序里。

而 Web 项目里，前端页面、手机 App、浏览器、第三方系统和你的 Python 后端不在同一个程序里，它们要通过网络请求来调用后端能力。

比如新闻 App：

前端页面：我要新闻列表
        ↓ HTTP 请求
FastAPI 后端：收到请求，查询数据库
        ↓
MySQL / Redis
        ↓
FastAPI 后端：整理成 JSON
        ↓ HTTP 响应
前端页面：展示新闻

FastAPI 的作用就是：

接收 HTTP 请求
解析请求参数
校验请求体
调用业务逻辑
查询数据库 / Redis / 大模型
返回 JSON 响应
自动生成接口文档

一句话理解：

FastAPI 是把 Python 函数发布成 Web API 的后端框架。

2. Web API 第一课：把 Python 函数变成网络接口

普通 Python 函数：

def get_news_list():
    return ["新闻1", "新闻2"]

只能在 Python 内部调用。

FastAPI 接口：

from fastapi import FastAPI

app = FastAPI()

@app.get("/news/list")
def get_news_list():
    return ["新闻1", "新闻2"]

浏览器访问：

http://127.0.0.1:8000/news/list

得到：

["新闻1", "新闻2"]

核心对应关系：

@app.get("/news/list")
def get_news_list():
    return {"data": "新闻列表"}

等价于：

浏览器 GET /news/list
        ↓
FastAPI 自动执行 get_news_list()
        ↓
函数 return 的字典变成 JSON
        ↓
返回给浏览器

3. HTTP 请求、JSON、Swagger 文档

3.1 常见 HTTP 方法

方法常见含义示例
GET获取数据获取新闻列表
POST新增数据 / 提交数据用户注册、登录
PUT修改数据修改用户信息
DELETE删除数据删除收藏、清空历史

注意：这只是常见约定，不是语法强制。但真实项目最好按这个规则写。

3.2 JSON 是前后端通信的主流格式

FastAPI 里返回 Python 字典：

return {"message": "Hello FastAPI"}

浏览器拿到的是 JSON：

{
  "message": "Hello FastAPI"
}

前端向后端提交注册数据，也是 JSON：

{
  "username": "admin",
  "password": "123456"
}

3.3 Swagger 自动接口文档

FastAPI 默认提供两个接口文档：

http://127.0.0.1:8000/docs
http://127.0.0.1:8000/redoc

学习阶段优先用 /docs，因为可以直接点 Try it out 测试接口。

4. FastAPI 最小项目

4.1 安装

pip install fastapi uvicorn

如果使用新版推荐安装方式：

pip install "fastapi[standard]"

4.2 创建 main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

4.3 启动

uvicorn main:app --reload

含义：

uvicorn       使用 Uvicorn 服务器启动项目
main:app      main.py 文件里的 app 对象
--reload      开发阶段自动重载

不要写成：

uvicorn main.py:app --reload

正确是：

uvicorn main:app --reload

5. AI 掘金头条项目结构

你的项目大致是：

FastAPI_first/
├── toutiao_backend/       # FastAPI 后端
└── xwzx-news/             # Vue/Vite 前端

后端：

toutiao_backend/
├── main.py
├── requirements.txt
├── test_main.http
├── cache/
│   └── news_cache.py
├── config/
│   ├── cache_conf.py
│   └── db_conf.py
├── crud/
│   ├── favorite.py
│   ├── history.py
│   ├── news.py
│   ├── news_cache.py
│   └── users.py
├── models/
│   ├── favorite.py
│   ├── history.py
│   ├── news.py
│   └── users.py
├── routers/
│   ├── favorite.py
│   ├── history.py
│   ├── news.py
│   └── users.py
├── schemas/
│   ├── base.py
│   ├── favorite.py
│   ├── history.py
│   ├── news.py
│   └── users.py
└── utils/
    ├── auth.py
    ├── exception.py
    ├── exception_handlers.py
    ├── response.py
    └── security.py

各目录职责：

目录作用
main.pyFastAPI 入口，创建 app，注册路由、中间件、异常处理器
routers/路由层，定义 URL、HTTP 方法、请求参数
crud/数据库操作层，封装查询、新增、修改、删除
models/SQLAlchemy ORM 模型，对应数据库表
schemas/Pydantic 模型，对应请求体和响应体
config/MySQL、Redis 等配置
utils/通用工具：认证、加密、响应、异常
cache/Redis 缓存操作

6. 后端 main.py 逐行拆解

典型 main.py：

from fastapi import FastAPI
from routers import news, users, favorite, history
from fastapi.middleware.cors import CORSMiddleware

from utils.exception_handlers import register_exception_handlers

app = FastAPI()

# 注册异常处理器
register_exception_handlers(app)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

app.include_router(news.router)
app.include_router(users.router)
app.include_router(favorite.router)
app.include_router(history.router)

6.1 app = FastAPI()

创建 FastAPI 应用对象。

以后所有接口、中间件、异常处理、文档生成，都是挂在这个 app 上。

6.2 register_exception_handlers(app)

注册全局异常处理器。作用是把不同类型的异常转成统一 JSON 响应。

6.3 app.add_middleware(CORSMiddleware, ...)

添加 CORS 跨域中间件。

开发阶段经常先写：

allow_origins=["*"]

但如果项目涉及 Authorization、Cookie、Bearer Token，更建议显式指定前端地址：

origins = [
    "http://localhost:5173",
    "http://127.0.0.1:5173",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization", "Accept", "Origin", "X-Requested-With"],
)

6.4 @app.get("/")

根接口，访问：

GET http://127.0.0.1:8000/

返回：

{"message": "Hello World"}

6.5 app.include_router(...)

把模块化路由挂到主应用。

例如：

app.include_router(users.router)

表示注册用户模块的所有接口。

7. APIRouter 模块化路由

如果所有接口都写在 main.py，项目会很乱。

错误倾向：

@app.get("/api/news/list")
async def get_news_list(): ...

@app.post("/api/user/login")
async def login(): ...

@app.get("/api/favorite/list")
async def favorite_list(): ...

接口一多，main.py 爆炸。

正确做法是把不同模块拆到不同文件。

例如 routers/users.py：

from fastapi import APIRouter

router = APIRouter(prefix="/api/user", tags=["users"])

@router.post("/register")
async def register():
    return {"message": "注册成功"}

完整路径：

POST /api/user/register

因为：

prefix = /api/user
局部路径 = /register
最终路径 = /api/user/register

tags=["users"] 用于 Swagger 文档分组。

8. Pydantic：请求体和响应体模型

8.1 请求体模型

注册接口需要前端传：

{
  "username": "admin",
  "password": "123456"
}

用 Pydantic 定义：

from pydantic import BaseModel

class UserRegisterRequest(BaseModel):
    username: str
    password: str

接口使用：

@router.post("/register")
async def register(user_data: UserRegisterRequest):
    username = user_data.username
    password = user_data.password

FastAPI 自动做：

读取 JSON 请求体
校验字段是否存在
校验类型是否正确
转换成 UserRegisterRequest 对象
注入到 user_data 参数

8.2 响应体模型

不要把数据库用户对象完整返回给前端，尤其不要返回密码。

定义响应模型：

from pydantic import BaseModel, ConfigDict

class UserInfoResponse(BaseModel):
    id: int
    username: str
    nickname: str | None = None
    avatar: str | None = None
    bio: str | None = None

    model_config = ConfigDict(from_attributes=True)

from_attributes=True 的意思是：可以从 ORM 对象属性读取数据。

例如：

UserInfoResponse.model_validate(user)

可以读取：

user.id
user.username
user.avatar

9. Depends：依赖注入

很多接口都需要重复逻辑：

获取数据库连接
验证 Token
获取当前用户
分页参数
权限校验

FastAPI 的 Depends 可以把这些公共逻辑抽出来。

9.1 数据库依赖

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

接口里使用：

@router.post("/register")
async def register(
    user_data: UserRegisterRequest,
    db: AsyncSession = Depends(get_db)
):
    ...

含义：

接口执行前：FastAPI 调用 get_db，拿到 session
接口执行中：db 参数就是数据库会话
接口执行后：成功 commit，失败 rollback，最后 close

9.2 当前用户依赖

async def get_current_user(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db)
):
    ...

接口里使用：

@router.get("/info")
async def get_user_info(user: User = Depends(get_current_user)):
    return success_response(data=user)

含义：

访问 /info 前，必须先通过 get_current_user 验证身份
验证通过，user 参数就是当前登录用户
验证失败，直接返回错误

10. SQLAlchemy 异步 ORM

本项目使用 SQLAlchemy 异步 ORM + MySQL。

典型配置：

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/news_app?charset=utf8mb4"

async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,
    pool_size=10,
    max_overflow=20,
)

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

10.1 连接字符串

mysql+aiomysql://root:123456@localhost:3306/news_app?charset=utf8mb4

含义：

片段含义
mysql数据库类型
aiomysql异步 MySQL 驱动
root用户名
123456密码
localhost数据库主机
3306MySQL 默认端口
news_app数据库名
charset=utf8mb4字符集

10.2 查询

result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()

解释：

select(User)                         查询 User 表
where(User.username == username)     加条件
await db.execute(...)                异步执行 SQL
scalar_one_or_none()                 取一条或 None

10.3 新增

user = User(username="admin", password="hashed")
db.add(user)
await db.flush()

db.add(user)：加入 session，准备插入。
flush()：把 SQL 先发给数据库，方便拿到自增 id，但还不是最终 commit。

10.4 删除

from sqlalchemy import delete

stmt = delete(Favorite).where(Favorite.user_id == user.id)
result = await db.execute(stmt)

10.5 分页

offset = (page - 1) * page_size
stmt = select(News).offset(offset).limit(page_size)

页码公式：

offset = (当前页码 - 1) * 每页数量

11. 用户模块完整拆解

用户模块一般包含：

POST /api/user/register       注册
POST /api/user/login          登录
GET  /api/user/info           获取当前用户信息
PUT  /api/user/update         修改用户信息
PUT  /api/user/password       修改密码

11.1 注册接口流程

前端提交 username/password
        ↓
Pydantic 校验请求体
        ↓
查询用户名是否已存在
        ↓
不存在则加密密码
        ↓
写入 user 表
        ↓
生成 token
        ↓
写入 user_token 表
        ↓
返回 token + userInfo

示例代码：

@router.post("/register")
async def register(
    user_data: UserRegisterRequest,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(User).where(User.username == user_data.username)
    )
    existing_user = result.scalar_one_or_none()

    if existing_user:
        raise BusinessException(message="用户名已存在")

    user = User(
        username=user_data.username,
        password=get_hash_password(user_data.password)
    )
    db.add(user)
    await db.flush()

    token = str(uuid.uuid4())
    user_token = UserToken(user_id=user.id, token=token)
    db.add(user_token)

    return success_response(
        message="注册成功",
        data={
            "token": token,
            "userInfo": UserInfoResponse.model_validate(user)
        }
    )

11.2 登录接口流程

前端提交 username/password
        ↓
查询用户是否存在
        ↓
验证明文密码和数据库哈希密码是否匹配
        ↓
生成 token
        ↓
写入 user_token 表
        ↓
返回 token + userInfo

密码验证：

pwd_context.verify(plain_password, hashed_password)

11.3 获取用户信息

@router.get("/info")
async def get_user_info(user: User = Depends(get_current_user)):
    return success_response(data=UserInfoResponse.model_validate(user))

用户必须在请求头携带：

Authorization: Bearer <token>

11.4 修改用户信息

验证 Token
    ↓
找到当前用户
    ↓
修改 nickname/avatar/gender/bio/phone
    ↓
返回更新后的用户信息

11.5 修改密码

验证 Token
    ↓
校验旧密码是否正确
    ↓
新密码加密
    ↓
更新 user.password
    ↓
返回密码修改成功

12. Token 和 Header 认证

HTTP 是无状态的。服务器天然不知道这次请求和上次登录是不是同一个人。

Token 的作用：

登录 / 注册成功后，后端生成一段 token 字符串
前端保存 token
以后每次请求个人接口时，把 token 放在请求头
后端根据 token 判断你是谁

请求头格式：

Authorization: Bearer <token>

含义：

部分含义
Authorization专门放认证信息的请求头
Bearer持有者令牌
<token>真正的身份凭证

示例：

Authorization: Bearer 550e8400-e29b-41d4-a716-446655440000

后端读取：

from fastapi import Header

async def get_current_user(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db)
):
    if not authorization.startswith("Bearer "):
        raise BusinessException(message="认证格式错误")

    token = authorization.replace("Bearer ", "")

    result = await db.execute(
        select(UserToken).where(UserToken.token == token)
    )
    user_token = result.scalar_one_or_none()

    if not user_token:
        raise BusinessException(message="登录已失效")

    user = await db.get(User, user_token.user_id)
    if not user:
        raise BusinessException(message="用户不存在")

    return user

13. passlib / bcrypt 密码加密

13.1 为什么不能明文存密码

错误：

username = admin
password = 123456

如果数据库泄露，用户密码全部暴露。

正确：

username = admin
password = $2b$12$...一长串哈希...

13.2 passlib 基本写法

from passlib.context import CryptContext

pwd_context = CryptContext(
    schemes=["bcrypt"],
    deprecated="auto"
)


def get_hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

解释：

代码含义
CryptContext密码加密配置中心
schemes=["bcrypt"]使用 bcrypt 算法
deprecated="auto"自动处理过期算法兼容
hash()明文密码变哈希
verify()校验明文和哈希是否匹配

13.3 本项目实际踩坑：bcrypt 5.0.0 兼容问题

你遇到的错误：

ValueError: password cannot be longer than 72 bytes, truncate manually if necessary

调用链：

routers/users.py
  ↓
crud/users.py
  ↓
utils/security.py
  ↓
pwd_context.hash(password)
  ↓
passlib / bcrypt 报错

你的密码很短，但仍报 72 bytes 错误，原因是 passlib==1.7.4 与新版 bcrypt==5.0.0 存在兼容问题。

修复：

pip uninstall bcrypt -y
pip install bcrypt==4.3.0

并在 requirements.txt 固定：

passlib[bcrypt]==1.7.4
bcrypt==4.3.0

验证：

python -c "from passlib.context import CryptContext; ctx=CryptContext(schemes=['bcrypt'], deprecated='auto'); print(ctx.hash('123456'))"

正常会输出：

$2b$12$......

14. 统一响应格式与全局异常处理

14.1 统一成功响应

推荐统一格式：

{
  "code": 200,
  "message": "success",
  "data": {}
}

封装函数：

from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse


def success_response(message: str = "success", data=None):
    content = {
        "code": 200,
        "message": message,
        "data": data,
    }
    return JSONResponse(content=jsonable_encoder(content))

jsonable_encoder() 的作用是把 Pydantic 模型、ORM 对象、datetime 等转换成 JSON 兼容数据。

14.2 业务异常

class BusinessException(Exception):
    def __init__(self, message: str = "业务异常", code: int = 400):
        self.message = message
        self.code = code

抛出：

raise BusinessException(message="用户名已存在")

14.3 全局异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse

async def business_exception_handler(request: Request, exc: BusinessException):
    return JSONResponse(
        status_code=200,
        content={
            "code": exc.code,
            "message": exc.message,
            "data": None,
        },
    )


def register_exception_handlers(app):
    app.add_exception_handler(BusinessException, business_exception_handler)

注册顺序建议：

业务异常
数据库约束异常
数据库异常
所有未知异常

15. 新闻模块：分类、列表、详情

新闻模块通常包含：

GET /api/news/categories
GET /api/news/list?categoryId=1&page=1&pageSize=10
GET /api/news/detail?id=1

15.1 获取分类

@router.get("/categories")
async def get_categories(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Category).offset(skip).limit(limit))
    categories = result.scalars().all()
    return success_response(data=categories)

15.2 获取新闻列表

参数：

categoryId 分类 id
page 当前页
pageSize 每页数量

分页公式：

offset = (page - 1) * page_size

是否还有更多：

has_more = offset + len(news_list) < total

15.3 获取新闻详情

@router.get("/detail")
async def get_news_detail(
    news_id: int = Query(..., alias="id"),
    db: AsyncSession = Depends(get_db)
):
    news = await db.get(News, news_id)
    if not news:
        raise BusinessException(message="新闻不存在")
    return success_response(data=news)

16. 收藏与浏览历史模块

16.1 收藏模块

接口：

GET    /api/favorite/check?newsId=1
POST   /api/favorite/add
DELETE /api/favorite/remove?newsId=1
GET    /api/favorite/list?page=1&pageSize=10
DELETE /api/favorite/clear

所有收藏接口一般都需要登录：

user: User = Depends(get_current_user)

检查收藏状态：

stmt = select(Favorite).where(
    Favorite.user_id == user.id,
    Favorite.news_id == news_id,
)
result = await db.execute(stmt)
favorite = result.scalar_one_or_none()

return success_response(data={"isFavorite": favorite is not None})

防止重复收藏，数据库层最好设置唯一约束：

UniqueConstraint("user_id", "news_id", name="user_news_unique")

16.2 浏览历史模块

接口：

POST   /api/history/add
GET    /api/history/list?page=1&pageSize=10
DELETE /api/history/delete/{history_id}
DELETE /api/history/clear

添加浏览历史时，推荐逻辑：

如果用户已经浏览过这篇新闻：更新浏览时间
否则：新增一条历史记录

注意本项目排查时发现一个潜在问题：

接口路径参数叫 history_id
但 CRUD 层可能按 news_id 删除

建议统一为：

async def delete_history(db: AsyncSession, user_id: int, history_id: int):
    stmt = delete(History).where(
        History.user_id == user_id,
        History.id == history_id,
    )
    result = await db.execute(stmt)
    return result.rowcount

17. Redis 缓存

Redis 是高性能 Key-Value 内存数据库，常用于缓存热点数据。

17.1 为什么要缓存

没有缓存：

1000 人查新闻分类 → 1000 次查 MySQL

有缓存：

第 1 次查 MySQL，写入 Redis
后面 999 次直接查 Redis

17.2 Cache-Aside 旁路缓存策略

读取：

先查缓存
有缓存 → 返回
无缓存 → 查数据库 → 写缓存 → 返回

写入 / 更新：

先更新数据库
再删除或更新缓存

17.3 Redis 配置

import redis.asyncio as redis

REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0

redis_client = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True,
)

17.4 缓存 key 设计

news:categories
news:list:{category_id}:{page}:{page_size}
news:detail:{news_id}

17.5 缓存时间建议

数据类型建议时间
分类、配置7200 秒
列表数据600 秒
详情数据1800 秒
验证码120 秒

原则：

数据越稳定，缓存越久
数据变化越快，缓存越短

18. CORS 跨域问题

你前端是：

http://localhost:5173

后端是：

http://127.0.0.1:8000

浏览器判断同源看三个东西：

协议
域名 / 主机
端口

所以这两个地址不同源，会触发跨域。

常见报错：

No 'Access-Control-Allow-Origin' header is present on the requested resource

FastAPI 配置：

from fastapi.middleware.cors import CORSMiddleware

origins = [
    "http://localhost:5173",
    "http://127.0.0.1:5173",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization", "Accept", "Origin", "X-Requested-With"],
)

注意：

allow_origins=["*"]
allow_credentials=True

开发阶段可能能跑，但涉及 Cookie、Authorization、Bearer Token 时，最好显式写前端 origin。

如果 Swagger /docs 里测试接口也 500，那就不是 CORS 问题，而是后端接口本身报错。

19. PyCharm + Windows 本地运行流程

19.1 后端运行

进入后端目录：

cd D:\workspace\code\python\FastAPI_first\toutiao_backend

创建虚拟环境：

python -m venv .venv

激活：

.\.venv\Scripts\activate

安装依赖：

pip install -r requirements.txt

如果 Windows 上 uvloop 安装失败，生成 Windows 专用依赖：

(Get-Content requirements.txt) | Where-Object {$_ -notmatch '^uvloop=='} | Set-Content requirements-win.txt
pip install -r requirements-win.txt

本项目密码加密兼容建议：

pip uninstall bcrypt -y
pip install bcrypt==4.3.0

启动后端：

uvicorn main:app --reload

打开：

http://127.0.0.1:8000/
http://127.0.0.1:8000/docs

19.2 PyCharm Run Configuration

不要直接运行 main.py。

配置：

Configuration 类型：Python
Module name：uvicorn
Parameters：main:app --reload
Working directory：D:\workspace\code\python\FastAPI_first\toutiao_backend
Python interpreter：选择 fastapi_env 或项目 .venv

19.3 数据库

数据库配置：

ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/news_app?charset=utf8mb4"

确保：

MySQL 已启动
数据库 news_app 已创建
课程 SQL 文件已导入
表包括 user、user_token、news、news_category、favorite、history 等

创建数据库：

CREATE DATABASE news_app DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

19.4 前端运行

进入前端：

cd D:\workspace\code\python\FastAPI_first\xwzx-news

安装：

npm install

启动：

npm run dev

前端默认访问后端：

http://127.0.0.1:8000

20. 本项目常见报错与修复

20.1 ModuleNotFoundError: No module named 'routers'

原因：工作目录不对。

解决：必须在 toutiao_backend 下启动：

cd toutiao_backend
uvicorn main:app --reload

20.2 导入了第三方 schemas 包

报错：

File "site-packages\schemas\__init__.py"
print "Schema violation ..."
SyntaxError: Missing parentheses in call to 'print'

原因：项目目录 schemas/ 缺少 __init__.py，Python 导入了环境里的第三方 schemas 包。

修复：

New-Item -ItemType File -Force .\schemas\__init__.py
New-Item -ItemType File -Force .\routers\__init__.py
New-Item -ItemType File -Force .\crud\__init__.py
New-Item -ItemType File -Force .\models\__init__.py
New-Item -ItemType File -Force .\config\__init__.py
New-Item -ItemType File -Force .\utils\__init__.py
New-Item -ItemType File -Force .\cache\__init__.py

验证：

python -c "import schemas; print(schemas.__file__)"

应该输出项目内路径，而不是 site-packages。

20.3 注册接口 500：bcrypt 兼容问题

报错：

ValueError: password cannot be longer than 72 bytes

修复：

pip uninstall bcrypt -y
pip install bcrypt==4.3.0

requirements.txt：

passlib[bcrypt]==1.7.4
bcrypt==4.3.0

20.4 CORS 报错

浏览器报：

No Access-Control-Allow-Origin header is present

先确认：

如果 /docs 里测试接口也 500 → 后端接口本身错误
如果 /docs 正常，前端报 CORS → 修改 main.py CORS 配置

20.5 Unknown database 'news_app'

原因：数据库没创建。

解决：

CREATE DATABASE news_app DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

20.6 Table 'news_app.xxx' doesn't exist

原因：数据库有了，但 SQL 表没有导入。

解决：导入课程 SQL 文件。

20.7 Redis 连接失败

报错类似：

Error connecting to localhost:6379

如果代码里捕获了异常，接口可能还能查数据库。开发阶段可以先不启动 Redis。

Docker 启动 Redis：

docker run --name teamx-redis -p 6379:6379 -d redis:7

测试：

redis-cli ping

返回：

PONG

21. Codex 调试提示词

21.1 项目总览

请你完整阅读当前 FastAPI 项目代码，但不要修改任何文件。
重点分析 toutiao_backend。
输出：
1. 项目结构
2. main.py 是否是入口
3. routers、crud、models、schemas、config、utils、cache 的职责
4. 已注册路由
5. 所有接口路径、方法、函数、是否需要登录
6. 数据库和 Redis 配置位置
7. 可能的启动风险

21.2 启动报错定位

下面是 uvicorn main:app --reload 的完整 Traceback。
请只根据 Traceback 定位问题，不要猜测。
要求：
1. 找到最后一行异常
2. 找到业务文件和行号
3. 判断是依赖、导入、数据库、Redis、Pydantic、SQLAlchemy 还是代码 bug
4. 给出最小修复方案
5. 不要重构项目

21.3 用户模块逐行讲解

请逐行解释用户模块代码，重点文件：
routers/users.py
crud/users.py
models/users.py
schemas/users.py
utils/auth.py
utils/security.py
utils/response.py
utils/exception_handlers.py

要求：
1. 找出注册、登录、获取用户信息、修改信息、修改密码接口
2. 解释每个函数参数含义
3. 解释 Depends、Header、AsyncSession、Pydantic schema
4. 解释密码加密和 Token 认证链路
5. 不要修改代码

21.4 最小修复要求

请根据错误诊断做最小必要修改。
要求：
1. 不重构项目结构
2. 不改接口路径
3. 不大面积改命名
4. 每改一个文件前说明原因
5. 修改后运行 uvicorn main:app --reload
6. 再测试 /docs 和目标接口

22. 推荐学习路线

阶段 1：FastAPI API 基础

目标：会写最基本接口。

学习内容：

FastAPI()
@app.get / @app.post
路径参数
查询参数
请求体
JSON 响应
/docs 文档

阶段 2：项目结构

目标：看懂真实后端项目。

学习内容：

main.py
APIRouter
routers
schemas
models
crud
utils
config

阶段 3：用户模块

目标：掌握后端最核心能力。

学习内容：

注册
登录
密码哈希
Token
Header
Depends(get_current_user)
修改用户信息
修改密码

阶段 4：数据库 ORM

目标：能自己写增删改查。

学习内容：

create_async_engine
AsyncSession
select
where
add
flush
delete
join
func.count
分页

阶段 5：缓存和联调

目标：让项目像真实系统。

学习内容：

Redis
Cache-Aside
CORS
前端 Axios
Swagger 调试
PyCharm 调试

阶段 6：工程化

目标：以后能写自己的项目。

学习内容：

.env 配置
Alembic 数据库迁移
pytest 测试
Docker 部署
日志
统一异常
权限系统

23. 参考资料

结语

这套课程真正要学的不是“记住某几行代码”，而是掌握一个后端项目的基本骨架：

FastAPI 接收请求
Pydantic 校验数据
Depends 注入公共能力
SQLAlchemy 操作数据库
passlib 保护密码
Token 维持登录状态
Redis 加速热点数据
统一响应和异常让前后端对接稳定

当你能把用户注册、登录、新闻列表、收藏、历史、Redis 缓存这几条链路全部跑通，你就已经具备了独立写一个中小型 FastAPI 后端项目的基础。