📚 全栈开发学习系列 · 阶段二（Web 全栈）
09 HTML 基础：网页骨架与语义化标签 ✓
10 CSS 基础：样式、布局与响应式 ✓
11 JavaScript 基础：变量、DOM 与事件 ✓
12 HTTP 协议基础：请求、响应与状态码 ✓
13 FastAPI 入门：构建第一个 RESTful API ✓
14 PostgreSQL 基础：数据库设计与 SQL 操作 ✓
15 认证授权：JWT、Session 与 Cookie ✓
16 综合实战：带登录的博客系统（当前篇）

综合实战：带登录的博客系统

难度：进阶 | 阶段二收官篇 · 整合 FastAPI + PostgreSQL + JWT
读完本篇你将能：把前几篇学到的 FastAPI 路由、PostgreSQL 数据持久化、JWT 认证三大模块组装成一个完整的博客后端系统——用户注册、登录、发文、编辑、删除，一条龙跑通。
📑 本文目录
01项目概览与架构设计
02项目结构与依赖安装
03数据库模型：User 与 Post
04认证模块：注册与登录
05博客文章 CRUD API
06接口联调与完整测试
07阶段收官与下一步规划

01 项目概览与架构设计

你已经分别掌握了 FastAPI 路由编写（第13篇）、PostgreSQL 数据操作（第14篇）和 JWT 认证（第15篇）。现在把这三块拼起来——构建一个真实的博客系统后端，包含用户注册、登录、发文章、编辑、删除的完整闭环。

这就好比盖房子：FastAPI 是钢筋骨架（路由与请求处理），PostgreSQL 是地基（数据持久存储），JWT 是门锁系统（身份验证）。三者缺一不可，组合起来才是一个能住人的房子。

系统架构一览

客户端请求
curl / Swagger
→
FastAPI
路由 + Pydantic
→
JWT 认证
Token 校验
→
PostgreSQL
SQLAlchemy

技术栈总览

层级 技术选型 版本
Web 框架 FastAPI 0.139.x
数据库 PostgreSQL 18
ORM SQLAlchemy 2.0 (async) 2.0.x
认证 PyJWT + pwdlib[argon2] 2.13.0 / 0.2.x
数据验证 Pydantic v2 2.x

功能清单

• 用户注册：POST /register，密码 Argon2 哈希存储
• 用户登录：POST /token，返回 JWT Token
• 发布文章：POST /posts，需登录
• 文章列表：GET /posts，公开访问，支持分页
• 编辑文章：PUT /posts/{id}，仅作者可操作
• 删除文章：DELETE /posts/{id}，仅作者可操作

02 项目结构与依赖安装

架构明确了，接下来搭骨架。一个清晰的项目结构能让代码各司其职，后期维护不混乱。我们把博客系统拆成五个模块：配置、数据库、模型、认证、路由。

目录结构

项目目录结构
blog/
├── main.py        # FastAPI 应用入口
├── config.py      # 配置管理（数据库URL、密钥等）
├── database.py     # 数据库引擎与 Session
├── models.py       # SQLAlchemy 数据模型
├── schemas.py      # Pydantic 请求/响应模型
├── auth.py        # 密码哈希 + JWT 签发/验证
├── dependencies.py  # 公共依赖（get_db、get_current_user）
└── .env          # 环境变量（不提交 Git）

安装依赖

一条命令装齐所有依赖：

Shell
pip install "fastapi[standard]" sqlalchemy[asyncio] \
    asyncpg pyjwt "pwdlib[argon2]" python-dotenv

各包的职责：

• fastapi[standard] — 含 uvicorn 服务器和 Pydantic v2
• sqlalchemy[asyncio] — SQLAlchemy 2.0 异步支持
• asyncpg — PostgreSQL 异步驱动
• pyjwt — JWT 签发与验证（2.13.0，2026-05 安全更新版）
• pwdlib[argon2] — 密码哈希（替代已停止维护的 passlib）

配置文件

把敏感信息放进 .env 文件，不硬编码在源码里：

.env
# PostgreSQL 连接串
DATABASE_URL=postgresql+asyncpg://bloguser:blogpass@localhost:5432/blogdb
# JWT 密钥（生产环境用 openssl rand -hex 32 生成）
SECRET_KEY=dev-secret-change-in-production
# Token 过期时间（小时）
ACCESS_TOKEN_EXPIRE_HOURS=24

然后在 config.py 中读取：

config.py
1
2
3
4
5
6
7
8
9
import os
from dotenv import load_dotenv
load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_HOURS = int(os.getenv("ACCESS_TOKEN_EXPIRE_HOURS", "24"))

03 数据库模型：User 与 Post

配置就绪后，先定义数据结构。博客系统只需两张表：users（用户表）和 posts（文章表），通过外键关联——一篇文章必定属于一个用户。

数据库引擎与 Session

database.py 负责创建异步引擎和 Session 工厂：

database.py — 异步引擎 + Session
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from config import DATABASE_URL
class Base(DeclarativeBase):
    """所有模型的基类"""
    pass
# 异步引擎：pool_pre_ping 防止连接断开后报错
engine = create_async_engine(DATABASE_URL, echo=True, pool_pre_ping=True)
# 异步 Session 工厂
async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
    """FastAPI 依赖：每个请求获取独立 Session"""
    async with async_session() as session:
        yield session
💡 小贴士
expire_on_commit=False 很关键：默认 commit 后 Session 中的对象会过期，异步环境下重新查询会报错。设为 False 让对象在 commit 后仍然可用。

SQLAlchemy 数据模型

两张表通过 user_id 外键关联，并设置 CASCADE 删除规则——用户被删时，其文章一并删除：

models.py — User 与 Post 模型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from datetime import datetime
from sqlalchemy import String, Text, ForeignKey, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from database import Base
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    hashed_password: Mapped[str] = mapped_column(String(255)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    # 一对多关系：一个用户可以有多篇文章
    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")
class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    published: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
    # 外键：指向 users.id，删除用户时级联删除文章
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    # 多对一关系：每篇文章属于一个用户
    author: Mapped["User"] = relationship(back_populates="posts")
async def init_db():
    """启动时自动建表（开发环境用，生产用 Alembic 迁移）"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Pydantic 请求/响应模型

Pydantic 模型负责接口数据的验证和序列化——进来的数据要校验，出去的数据要过滤。绝不能把 hashed_password 返回给客户端：

schemas.py — Pydantic 模型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, ConfigDict
class UserCreate(BaseModel):
    """注册请求"""
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=6)
class UserOut(BaseModel):
    """用户响应（不含密码）"""
    model_config = ConfigDict(from_attributes=True)
    id: int
    username: str
    email: str
    created_at: datetime
class PostCreate(BaseModel):
    """创建文章请求"""
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=1)
    published: bool = True
class PostOut(BaseModel):
    """文章响应"""
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    content: str
    published: bool
    created_at: datetime
    updated_at: datetime
    author_id: int
⚠️ 常见错误
UserOut 包含 hashed_password 字段 — 响应模型中绝不能出现密码字段，哪怕是哈希后的
✓ 正确：UserOut 只暴露 id、username、email、created_at 四个安全字段。model_config = ConfigDict(from_attributes=True) 让 Pydantic 自动从 SQLAlchemy 对象读取属性。

04 认证模块：注册与登录

数据库表建好了，接下来实现安全认证。这里复用第15篇的知识：pwdlib + Argon2 做密码哈希，PyJWT 签发 Token。把认证逻辑独立到 auth.py 中，保持代码整洁。

密码哈希与 JWT 工具函数

auth.py — 密码哈希 + JWT 工具
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import jwt
from datetime import datetime, timedelta, timezone
from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher
from config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_HOURS
# 创建 Argon2 哈希器实例
pwd_hasher = PasswordHash((Argon2Hasher(),))
def hash_password(password: str) -> str:
    """注册时调用：明文 → Argon2 哈希"""
    return pwd_hasher.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
    """登录时调用：验证明文与哈希是否匹配"""
    return pwd_hasher.verify(plain, hashed)
def create_access_token(data: dict) -> str:
    """签发 JWT：把用户信息编码为带签名的 Token"""
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict:
    """验证 JWT：检查签名和过期时间，返回 Payload"""
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
# OAuth2 Token 提取器：从 Authorization: Bearer xxx 头中取 Token
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

注册接口

注册流程：接收用户名/邮箱/密码 → 检查是否已存在 → 密码哈希 → 存入数据库 → 返回用户信息（不含密码）。

main.py — 注册接口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from models import User
from schemas import UserCreate, UserOut
from auth import hash_password
app = FastAPI(title="博客系统 API")
@app.post("/register", response_model=UserOut, status_code=201)
async def register(user: UserCreate, db: AsyncSession = Depends(get_db)):
    # 检查用户名是否已存在
    result = await db.execute(select(User).where(User.username == user.username))
    if result.scalar_one_or_none():
        raise HTTPException(400, "用户名已被注册")
    # 创建用户：密码哈希后存储
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hash_password(user.password),
    )
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

登录接口

登录流程：接收 OAuth2 表单格式的用户名/密码 → 查库验证 → 签发 JWT Token。使用 OAuth2PasswordRequestForm 保证 Swagger UI 的"Authorize"按钮可用。

main.py — 登录接口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from fastapi.security import OAuth2PasswordRequestForm
from auth import verify_password, create_access_token
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends(),
               db: AsyncSession = Depends(get_db)):
    # 按用户名查库
    result = await db.execute(select(User).where(User.username == form_data.username))
    user = result.scalar_one_or_none()
    # 用户不存在或密码错误 → 统一返回 401（不泄露用户是否存在）
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 签发 Token：sub（subject）是 JWT 标准声明，存用户名
    token = create_access_token(data={"sub": user.username, "uid": user.id})
    return {"access_token": token, "token_type": "bearer"}
💡 小贴士
登录失败时返回 "用户名或密码错误"而非"用户不存在"——这是安全实践，防止攻击者通过接口枚举注册用户。sub 是 JWT 标准声明字段（RFC 7519），表示 Token 的主体，这里存用户名。

当前用户依赖

把"从 Token 提取当前用户"封装为公共依赖，所有需要登录的接口只需声明 Depends(get_current_user) 即可：

dependencies.py — 当前用户依赖
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from auth import decode_access_token, oauth2_scheme
from database import get_db
from models import User
async def get_current_user(token: str = Depends(oauth2_scheme),
                           db: AsyncSession = Depends(get_db)) -> User:
    credentials_exc = HTTPException(401, "无法验证凭据", headers={"WWW-Authenticate": "Bearer"})
    try:
        payload = decode_access_token(token)
        username: str = payload.get("sub")
    except (jwt.InvalidTokenError, jwt.ExpiredSignatureError):
        raise credentials_exc
    result = await db.execute(select(User).where(User.username == username))
    user = result.scalar_one_or_none()
    if not user:
        raise credentials_exc
    return user

05 博客文章 CRUD API

认证模块就绪，现在实现博客核心功能——文章的增删改查。关键设计：创建/编辑/删除需要登录，且编辑和删除只能操作自己的文章；列表和详情公开访问。

发布文章（需登录）

main.py — 发布文章
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from models import Post
from schemas import PostCreate, PostOut
from dependencies import get_current_user
@app.post("/posts", response_model=PostOut, status_code=201)
async def create_post(post: PostCreate,
                      db: AsyncSession = Depends(get_db),
                      current_user: User = Depends(get_current_user)):
    db_post = Post(title=post.title, content=post.content,
                  published=post.published, user_id=current_user.id)
    db.add(db_post)
    await db.commit()
    await db.refresh(db_post)
    return db_post

文章列表与详情（公开）

main.py — 列表与详情
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@app.get("/posts", response_model=list[PostOut])
async def list_posts(skip: int = 0, limit: int = 10,
                 db: AsyncSession = Depends(get_db)):
    # 只返回已发布的文章，按创建时间倒序
    result = await db.execute(
        select(Post).where(Post.published == True)
        .order_by(Post.created_at.desc()).offset(skip).limit(limit)
    )
    return result.scalars().all()
@app.get("/posts/{post_id}", response_model=PostOut)
async def get_post(post_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Post).where(Post.id == post_id))
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(404, "文章不存在")
    return post

编辑与删除（仅作者可操作）

编辑和删除接口需要双重校验：先验证登录状态，再验证当前用户是否是文章作者。这通过 post.user_id != current_user.id 判断实现：

main.py — 编辑与删除
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from schemas import PostCreate
@app.put("/posts/{post_id}", response_model=PostOut)
async def update_post(post_id: int, post_update: PostCreate,
                        db: AsyncSession = Depends(get_db),
                        current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Post).where(Post.id == post_id))
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(404, "文章不存在")
    if post.user_id != current_user.id:
        raise HTTPException(403, "只能编辑自己的文章")
    post.title = post_update.title
    post.content = post_update.content
    post.published = post_update.published
    await db.commit()
    await db.refresh(post)
    return post
@app.delete("/posts/{post_id}", status_code=204)
async def delete_post(post_id: int,
                     db: AsyncSession = Depends(get_db),
                     current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Post).where(Post.id == post_id))
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(404, "文章不存在")
    if post.user_id != current_user.id:
        raise HTTPException(403, "只能删除自己的文章")
    await db.delete(post)
    await db.commit()
⚠️ 常见错误
忘记检查 post.user_id != current_user.id — 任何登录用户都能编辑/删除别人的文章，这是严重的越权漏洞
✓ 正确：先查 404（文章不存在），再查 403（不是作者），最后执行操作。403 和 404 的顺序很重要——先确认资源存在再检查权限，避免通过 403 泄露文章是否存在。

接口权限总览

接口 方法 认证 额外限制
/register POST 公开 用户名唯一
/token POST 公开 返回 JWT
/posts GET 公开 支持分页
/posts/{id} GET 公开 —
/posts POST 需登录 —
/posts/{id} PUT 需登录 仅作者
/posts/{id} DELETE 需登录 仅作者

06 接口联调与完整测试

所有接口代码写完了，现在启动服务、建库建表、用 curl 走一遍完整流程：注册 → 登录拿 Token → 发文章 → 查列表 → 编辑 → 删除。

启动服务

先在 PostgreSQL 中创建数据库和用户，然后启动 FastAPI：

Shell — 建库 + 启动
# 1. 创建数据库和用户
psql -U postgres -c "CREATE USER bloguser WITH PASSWORD 'blogpass';"
psql -U postgres -c "CREATE DATABASE blogdb OWNER bloguser;"
# 2. 启动 FastAPI（自动建表）
fastapi dev main.py
# 输出：
# INFO: Uvicorn running on http://127.0.0.1:8000
# INFO: Application startup complete.

完整联调流程

用 curl 逐步测试每个接口：

Shell — 完整联调
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 步骤1：注册用户
curl -X POST http://localhost:8000/register \
  -H "Content-Type: application/json" \
  -d '{"username":"alice","email":"alice@test.com","password":"secret123"}'
# → {"id":1,"username":"alice","email":"alice@test.com","created_at":"2026-08-13T..."}
# 步骤2：登录拿 Token
curl -X POST http://localhost:8000/token \
  -d "username=alice&password=secret123"
# → {"access_token":"eyJhbG...","token_type":"bearer"}
# 步骤3：发文章（带 Token）
curl -X POST http://localhost:8000/posts \
  -H "Authorization: Bearer eyJhbG..." \
  -H "Content-Type: application/json" \
  -d '{"title":"我的第一篇博客","content":"Hello World!","published":true}'
# → {"id":1,"title":"我的第一篇博客","content":"Hello World!",...}
# 步骤4：查文章列表（无需 Token）
curl http://localhost:8000/posts?skip=0&limit=10
# → [{"id":1,"title":"我的第一篇博客",...}]
# 步骤5：删除文章（仅作者）
curl -X DELETE http://localhost:8000/posts/1 \
  -H "Authorization: Bearer eyJhbG..."
# → 204 No Content（删除成功）

Swagger UI 在线测试

除了 curl，FastAPI 自带的 Swagger UI 是更直观的测试工具。打开浏览器访问 http://localhost:8000/docs，点击右上角"Authorize"按钮，输入用户名密码登录后，所有需要认证的接口都可以直接在页面上测试。

注册用户
→
登录获取 Token
→
Authorize 填入 Token
→
测试全部接口
💡 小贴士
生产环境务必做三项加固：1）用 openssl rand -hex 32 生成强密钥替换 .env 中的 SECRET_KEY；2）用 Alembic 管理数据库迁移而非 create_all 自动建表；3）添加 CORS 中间件允许前端跨域访问：app.add_middleware(CORSMiddleware, ...)。

07 阶段收官与下一步规划

恭喜！你刚刚从零搭建了一个完整的博客后端系统。这不是玩具代码——它使用了异步数据库引擎、Argon2 密码哈希、JWT 无状态认证、Pydantic v2 数据验证，每一项都是生产级技术选型。

本篇核心回顾

1 项目分层：config / database / models / schemas / auth / dependencies / main 七个模块各司其职
2 SQLAlchemy 2.0 异步模型：Mapped 类型注解 + mapped_column + relationship 一对多关联
3 认证闭环：pwdlib Argon2 哈希 + PyJWT 签发 + OAuth2PasswordBearer 提取 + Depends 注入
4 权限控制：公开接口（列表/详情）vs 登录接口（发文）vs 作者专属（编辑/删除），403/404 顺序有讲究

阶段二成果总结

从第09篇到第16篇，你走完了 Web 全栈的核心路径：

篇目 核心技能
09 HTML 语义化标签、表单、表格
10 CSS 选择器、盒模型、Flex 布局、响应式
11 JavaScript 变量、DOM 操作、事件处理
12 HTTP 请求/响应结构、状态码、RESTful
13 FastAPI 路由、Pydantic 验证、Swagger UI
14 PostgreSQL SQL 操作、表设计、SQLAlchemy 集成
15 认证授权 JWT、Session、Cookie、Argon2 哈希
16 综合实战 整合三者构建完整博客系统
📖 知识回顾
SQLAlchemy 异步模型 Pydantic 数据验证 Argon2 密码哈希 JWT 签发与验证 OAuth2PasswordBearer FastAPI 依赖注入 越权防护 403/404 CASCADE 级联删除 Swagger UI 联调
✏️ 动手练习
🟢 基础验证
在本篇博客系统基础上，添加 GET /users/me 接口，返回当前登录用户的信息。提示：复用 get_current_user 依赖，response_model 用 UserOut。
🟡 组合应用
为文章列表添加搜索功能：GET /posts?search=关键词，按标题模糊匹配。提示：SQLAlchemy 中用 Post.title.contains(search) 或 ilike() 方法。
🔴 开放挑战
为博客系统添加评论功能：新建 comments 表（id, content, post_id, user_id, created_at），实现 POST /posts/{id}/comments（需登录）和 GET /posts/{id}/comments（公开）两个接口。需要新建 Comment 模型、CommentCreate/CommentOut schema，并配置 Post-Comment 一对多关系。
阶段二 完结

HTML / CSS / JavaScript / HTTP / FastAPI / PostgreSQL / JWT / 综合实战 — 8 篇全部完成

下篇预告 · 阶段三 前端深化
17 React 入门：组件、Props 与 State
将从前端三剑客迈入现代前端框架——学习 React 组件化思维、JSX 语法、状态管理，为博客系统构建前端界面
关注公众号获取更多内容