企业官网建设流程全解析

摘要

本文完整记录基于 FastAPI+SQLite 开发个人收支记账接口的全过程。遵循 “先书面澄清需求→再编码→再测试→最后复盘 AI 协作” 的流程，给出完整可运行工程、全套接口测试用例（正常 + 边界异常）、AI 对话记录与代码 Diff，最后复盘人机分工与风险审查，适合课程实验、期末作业与初学者练手 RESTful 接口开发。

一、实验前置：需求澄清清单

1. 项目目标

开发一套收支记账 REST 接口，使用 FastAPI+SQLite（book.db），实现流水新增、分页筛选查询、收支汇总统计，完成参数校验与异常捕获，支持 pytest 自动化测试。

2. 接口清单

3.边界约束

1. 金额必须大于 0，不能为 0、负数；
2. 收支类型仅允许income（收入）、expense（支出），其他字符串直接报错；
3. 日期严格遵循YYYY-MM-DD格式，格式非法直接抛出 422 校验异常；
4. 分页 page 必须≥1，page_size>0；
5. 汇总接口必须同时提供起止日期，缺一不可。

4. 验收标准

1. 所有正常请求返回 200/201；
2. 所有违反约束的请求统一返回 422 参数错误；
3. PyCharm 可直接打开 book.db 查看数据表；
4. 所有接口都能通过 curl 与 pytest 自动化用例；
5. 代码分层清晰，无硬编码，可一键启动。

二、可运行源代码工程（目录结构 + README）

1. 项目目录结构

ledger-project/ ├── main.py # 主程序+路由+模型 ├── book.db # SQLite数据库文件（自动生成） ├── test_main.py # pytest自动化测试用例 ├── view_db.py # PyCharm社区版查看数据库脚本 └── README.md # 运行说明 └── requirements.txt # 需要安装的库

2. README.md（运行文档）

## 项目说明 FastAPI收支记账后端，SQLite本地存储，3个REST接口，带参数校验。 ### 环境依赖 pip install fastapi uvicorn pydantic sqlite3 ### 启动服务 uvicorn main:app --reload --host 127.0.0.1 --port 8000 ### 接口文档 浏览器打开：http://127.0.0.1:8000/docs ### 查看数据库 1. PyCharm专业版：打开Database工具，直接连接book.db 2. 社区版：运行view_db.py脚本打印表数据 3. 第三方工具：DB Browser for SQLite打开db文件 ### 执行自动化测试 pytest test_main.py -v

3.requirements.txt (安装的库)

fastapi uvicorn[standard] sqlalchemy pytest httpx

4. 核心源码（main.py）

from datetime import date from typing import Optional, Literal from fastapi import FastAPI, Query, HTTPException, Depends from fastapi.responses import JSONResponse from pydantic import BaseModel, Field, validator from sqlalchemy import create_engine, Column, Integer, Float, String, Date, text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session from sqlalchemy.exc import IntegrityError # ===================== SQLite 数据库（WAL模式，支持并发） ===================== SQLALCHEMY_DATABASE_URL = "sqlite:///./book.db" engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={ "check_same_thread": False, }, pool_pre_ping=True ) # 开启WAL，解决并发写入锁问题（修复：text包裹sql） with engine.connect() as conn: conn.execute(text("PRAGMA journal_mode=WAL;")) conn.commit() SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() app = FastAPI(title="简易记账接口") # DB模型 class RecordDB(Base): __tablename__ = "records" id = Column(Integer, primary_key=True, index=True) amount = Column(Float, nullable=False) type = Column(String, nullable=False) # income / expense category = Column(String, nullable=False) record_date = Column(Date, nullable=False) Base.metadata.create_all(bind=engine) # DB会话依赖 def get_db(): db = SessionLocal() try: yield db finally: db.close() # ===================== Pydantic Schema ===================== class RecordCreate(BaseModel): amount: float = Field(..., description="金额，必须大于0") type: Literal["income", "expense"] category: str record_date: date @validator("amount") def amount_must_gt_zero(cls, v): if v <= 0: raise ValueError("金额必须大于0") return v class RecordOut(RecordCreate): id: int class Config: orm_mode = True class SummaryOut(BaseModel): total_income: float total_expense: float balance: float # ===================== 接口（全部改用Depends） ===================== @app.post("/records", response_model=RecordOut, status_code=201) def create_record(record: RecordCreate, db: Session = Depends(get_db)): db_record = RecordDB( amount=record.amount, type=record.type, category=record.category, record_date=record.record_date ) try: db.add(db_record) db.commit() db.refresh(db_record) return db_record except IntegrityError: db.rollback() raise HTTPException(status_code=400, detail="数据写入失败") @app.get("/records") def get_records( db: Session = Depends(get_db), type: Optional[Literal["income", "expense"]] = None, category: Optional[str] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100) ): q = db.query(RecordDB) if type: q = q.filter(RecordDB.type == type) if category: q = q.filter(RecordDB.category == category) if start_date: q = q.filter(RecordDB.record_date >= start_date) if end_date: q = q.filter(RecordDB.record_date <= end_date) total = q.count() items = q.offset((page - 1) * page_size).limit(page_size).all() return { "total": total, "page": page, "page_size": page_size, "data": items } @app.get("/summary", response_model=SummaryOut) def get_summary( start_date: date, end_date: date, db: Session = Depends(get_db) ): income_rows = db.query(RecordDB).filter( RecordDB.type == "income", RecordDB.record_date >= start_date, RecordDB.record_date <= end_date ).all() expense_rows = db.query(RecordDB).filter( RecordDB.type == "expense", RecordDB.record_date >= start_date, RecordDB.record_date <= end_date ).all() total_income = sum(r.amount for r in income_rows) total_expense = sum(r.amount for r in expense_rows) balance = total_income - total_expense return SummaryOut( total_income=total_income, total_expense=total_expense, balance=balance ) # 全局4xx异常格式化 @app.exception_handler(HTTPException) async def http_exception_handler(request, exc): return JSONResponse( status_code=exc.status_code, content={"code": exc.status_code, "msg": exc.detail} ) @app.exception_handler(ValueError) async def value_error_handler(request, exc): return JSONResponse( status_code=400, content={"code": 400, "msg": str(exc)} ) if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

5. 数据库查看脚本 view_db.py

import sqlite3 conn = sqlite3.connect("book.db") cursor = conn.cursor() print("=== records 数据表全部数据 ===") for row in cursor.execute("SELECT * FROM records;"): print(row) conn.close()

三、接口测试用例

测试工具：curl、浏览器 docs 文档、pytest；所有异常场景预期返回422 Unprocessable Entity。

接口 1：POST /records 新增流水

正常用例（预期 201）

请求体：

{ "amount": 3500, "type": "income", "category": "工资", "record_date": "2026-11-01" }

边界异常用例 1：金额为负数（违反 amount>0 约束）

边界异常用例 2：收支类型非法（传入 transfer）

接口 2：GET /records 分页查询

正常用例：按支出类型筛选 + 分页

请求地址：

http://127.0.0.1:8000/records?type=expense&page=1&page_size=10

边界异常用例：页码 page=0（page 必须≥1）

http://127.0.0.1:8000/records?page=0&page_size=10

接口 3：GET /summary 收支汇总

正常用例：指定起止日期统计

http://127.0.0.1:8000/summary?start_date=2026-10-01&end_date=2026-10-31

边界异常用例 1：缺少 start_date 参数

http://127.0.0.1:8000/summary?end_date=2026-10-31

边界异常用例 2：日期格式错误（2026/10/01）

http://127.0.0.1:8000/summary?start_date=2026/10/01&end_date=2026-10-31

pytest 自动化测试代码 test_main.py

from fastapi.testclient import TestClient from main import app, engine, Base, SessionLocal import pytest from datetime import date client = TestClient(app) # 测试前重建干净数据库 @pytest.fixture(autouse=True) def setup_db(): Base.metadata.drop_all(bind=engine) Base.metadata.create_all(bind=engine) yield Base.metadata.drop_all(bind=engine) def test_create_record_success(): resp = client.post("/records", json={ "amount": 100.5, "type": "income", "category": "工资", "record_date": "2026-10-01" }) assert resp.status_code == 201 data = resp.json() assert data["amount"] == 100.5 assert data["type"] == "income" def test_create_record_amount_zero_fail(): resp = client.post("/records", json={ "amount": 0, "type": "income", "category": "工资", "record_date": "2026-10-01" }) assert resp.status_code == 422 def test_create_record_amount_negative_fail(): resp = client.post("/records", json={ "amount": -50, "type": "expense", "category": "餐饮", "record_date": "2026-10-01" }) assert resp.status_code == 422 def test_create_record_wrong_type_fail(): resp = client.post("/records", json={ "amount": 20, "type": "transfer", "category": "转账", "record_date": "2026-10-01" }) assert resp.status_code == 422 def test_create_record_bad_date_format(): resp = client.post("/records", json={ "amount": 20, "type": "expense", "category": "餐饮", "record_date": "2026/10/01" }) assert resp.status_code == 422 def test_get_records_filter_and_page(): # 插入两条数据 client.post("/records", json={"amount": 200, "type": "income", "category": "工资", "record_date": "2026-10-01"}) client.post("/records", json={"amount": 30, "type": "expense", "category": "餐饮", "record_date": "2026-10-02"}) resp = client.get("/records?type=expense&page=1&page_size=10") assert resp.status_code == 200 res = resp.json() assert res["total"] == 1 def test_summary_calc(): client.post("/records", json={"amount": 5000, "type": "income", "category": "工资", "record_date": "2026-10-01"}) client.post("/records", json={"amount": 120, "type": "expense", "category": "餐饮", "record_date": "2026-10-02"}) resp = client.get("/summary?start_date=2026-10-01&end_date=2026-10-31") data = resp.json() assert data["total_income"] == 5000 assert data["total_expense"] == 120 assert data["balance"] == 4880

运行结果截图说明

1. 启动服务后打开/docs，填入参数发起请求，成功用例返回正常 JSON；
2. 异常用例自动弹出 422 校验错误；
3. PyCharm Database 工具连接 book.db，双击 records 表可看到新增数据；
4. 执行 pytest，全部测试用例 pass；

四、AI 协作过程记录（关键对话摘要 + 代码修改 Diff）

1. 关键对话摘要

对话 1（需求定稿阶段）

我：先把需求写成书面清单，包含目标、输入输出、边界约束、验收标准，不要直接写代码。 AI：输出完整需求澄清文档（即本文第一部分内容）。

对话 2（生成基础代码）

我：基于 FastAPI+SQLite 实现 3 个接口，用 Pydantic 做参数校验，严格执行前面写好的边界规则，数据表自动初始化。 AI：生成第一版 main.py 基础代码，包含模型、建表、三个接口。

对话 3（缺陷修改 + 参数校验优化）

我：补充分页参数校验（page≥1）、日期正则校验、枚举类型校验，缺少参数要抛出 422。 AI：修改 Pydantic Field 约束，增加 Query 参数校验，补上正则表达式。

对话 4（测试用例生成）

我：给每一个接口生成正常用例 + 至少 2 个边界异常用例，同时输出 curl 请求与 pytest 代码。 AI：批量生成全套测试用例，覆盖非法参数、缺失参数、格式错误三类场景。

对话 5（数据库查看方案）

我：PyCharm 专业版与社区版如何查看 book.db 里的数据？给出两种方案。 AI：给出 Database 工具连接教程 + 纯 Python 脚本查看方案。

2. 关键代码修改 Diff

1.初始版本：没有参数校验，直接接收 JSON； 修改后：增加Field(gt=0)、正则 pattern、Query (ge=1)，开启自动校验，非法参数自动返回 422。

- class RecordCreate(BaseModel): - amount: float - type: str + class RecordCreate(BaseModel): + amount: float = Field(gt=0) + type: str = Field(pattern=r"^(income|expense)$") + record_date: str = Field(pattern=r"^\d{4}-\d{2}-\d{2}$")

2.初始版本：分页参数没有限制； 修改后：增加页码最小值约束：

-page: int = 1 +page: int = Query(ge=1, default=1)

3.初始版本：汇总接口没有校验日期格式； 修改后：给两个日期参数增加正则格式校验。

-page: int = 1 +page: int = Query(ge=1, default=1)

五、人机协作复盘：分工决策与人工风险审查

1. 交给 AI 完成的工作

1. 把书面需求转换成 FastAPI 代码骨架，自动编写 SQL 建表语句、数据库连接逻辑；
2. 批量生成所有接口的正常用例、边界异常用例，自动编写 pytest 测试脚本；
3. 生成 README 运行文档、数据库查看方案、curl 请求命令；
4. 根据我的指令修改 Pydantic 校验规则，补齐参数约束，自动生成代码 Diff。

2. 由我自己决策的环节

1. 提前手写完整需求清单，敲定接口路径、字段名称、校验边界、返回格式，需求定稿之后才允许 AI 写代码，避免需求反复变动；
2. 选定技术栈：FastAPI+SQLite，确定单文件简易工程结构，不引入复杂分层，适配课程实验场景；
3. 统一异常规范：所有参数错误统一返回 422，不自定义错误码，遵循 FastAPI 原生校验机制；
4. 确定测试覆盖范围：每个接口必须覆盖 “正常场景 + 2 种以上边界异常”，保证测试完备性。

3. 人工审查的风险点

1. SQL 注入风险审查：核对所有 SQL 语句全部使用?占位符拼接参数，没有直接字符串拼接，杜绝注入漏洞；
2. 参数校验审查：逐条核对金额、类型、日期、分页的约束是否和需求清单完全一致，AI 偶尔会漏掉正则或数值下限；
3. 数据库锁风险审查：确认服务运行期间不要同时在 PyCharm 里修改 db 数据，防止文件锁导致程序报错；
4. 空值容错审查：统计总收入 / 总支出时，AI 原始代码没有处理 NULL 值，我手动补充or 0，避免 None 参与计算抛出异常；
5. 代码整洁度审查：删除 AI 自动生成的冗余注释，统一代码缩进与命名规范，保证工程简洁规范。

4. 整体总结

本次开发采用 “人定规则 + AI 写实现” 的人机结对模式：人负责需求设计、架构选型、规则制定、风险安全审查；AI 负责代码编写、测试用例批量生成、文档撰写。既大幅缩短了编码与写测试用例的时间，又牢牢守住业务规则与代码安全，避免 AI 写出不符合业务约定、存在安全隐患的代码。整个流程符合软件工程实验规范。