一、实验目的：
- 使用 FastAPI 实现简易记账 HTTP 服务，掌握请求体、查询参数和 JSON 响应的处理方法。
- 完成流水新增、条件筛选、分页查询和收支汇总，理解 SQLite 持久化与事务处理。
- 使用 pytest 验证正常功能、非法参数与并发写入，并记录 AI 辅助编码的需求拆解、实现和验证过程。
二、实验环境：
| 项目 | 实验环境 |
|---|---|
| 操作系统 | macOS |
| Python | 3.12.14 |
| Web 框架 | FastAPI 0.146.0 |
| 服务程序 | Uvicorn 0.54.0 |
| 数据库 | SQLite 3.53.1 |
| 测试工具 | pytest 9.1.1、HTTPX 0.28.1 |
| 数据保存方式 | 本地 SQLite 文件，金额以整数“分”存储 |
运行方式：安装requirements.txt中的依赖后，在工程目录执行以下命令。详细说明见同目录README.md。
python-muvicorn main:app--host127.0.0.1--port8000python-mpytest-v三、实验内容：
1. 需求分析与验收约定
本实验只完成考题二。输入为 HTTP 请求中的 JSON 数据与查询参数，输出为 JSON 格式的流水、分页结果、收支汇总或错误信息。
| 接口 | 功能 | 关键规则 |
|---|---|---|
POST /records | 新增流水 | 金额大于 0、最多两位小数；类型为收入或支出；分类非空；日期为 YYYY-MM-DD；成功返回 201 |
GET /records | 查询流水 | 可组合筛选类型、分类、起止日期，支持 page 和 page_size，成功返回 200 |
GET /summary | 汇总收支 | 按指定日期区间计算收入、支出和结余，成功返回 200 |
| 非法请求 | 错误处理 | 参数校验失败返回 422，errors 数组给出字段、原因和错误类型 |
日期区间包含开始和结束当天，省略某一端表示不限制该端。开始日期晚于结束日期时拒绝请求。分页默认第 1 页，每页 10 条，最大每页 100 条；超出数据范围返回空列表。查询结果按日期降序、编号降序排列。无数据时汇总返回 0.00。
金额字段名为amount，类型字段为type，分类字段为category，日期字段为date。为避免金额计算时的二进制浮点误差，建议输入金额字符串；服务用 Decimal 校验后转换为整数分保存，返回时统一格式化为两位小数字符串。
2. 实现过程与主要代码
先定义输入模型和校验规则，再创建数据库表，随后实现新增、筛选分页和汇总，最后运行测试与真实 HTTP 调用。下列代码为工程中完整的main.py，代码中不含注释。
importosimportsqlite3fromcontextlibimportclosingfromdatetimeimportdatefromdecimalimportDecimalfrompathlibimportPathfromtypingimportAnnotated,LiteralfromfastapiimportDepends,FastAPI,Queryfromfastapi.exceptionsimportRequestValidationErrorfromfastapi.responsesimportJSONResponsefrompydanticimportBaseModel,ConfigDict,Field,field_validator RecordType=Literal["收入","支出"]classRecordInput(BaseModel):model_config=ConfigDict(extra="forbid")amount:Decimal=Field(gt=0,max_digits=12,decimal_places=2)type:RecordType category:str=Field(min_length=1,max_length=50)date:date@field_validator("date",mode="before")@classmethoddefvalidate_date(cls,value):ifnotisinstance(value,str):raiseValueError("日期必须使用 YYYY-MM-DD 格式")parsed=date.fromisoformat(value)ifparsed.isoformat()!=value:raiseValueError("日期必须使用 YYYY-MM-DD 格式")returnvalue@field_validator("category")@classmethoddefvalidate_category(cls,value):value=value.strip()ifnotvalue:raiseValueError("分类不能为空")returnvaluedefdate_range(start_date:date|None=None,end_date:date|None=None,):ifstart_dateandend_dateandstart_date>end_date:raiseRequestValidationError([{"loc":["query","end_date"],"msg":"结束日期不能早于开始日期","type":"date_range",}])returnstart_date,end_datedefcreate_app(db_path=None):db_path=str(db_pathoros.getenv("ACCOUNT_DB","records.db"))Path(db_path).parent.mkdir(parents=True,exist_ok=True)defconnect():connection=sqlite3.connect(db_path,timeout=10)connection.row_factory=sqlite3.Rowreturnconnectionwithclosing(connect())asconnection:connection.execute("PRAGMA journal_mode=WAL")connection.execute(""" CREATE TABLE IF NOT EXISTS records ( id INTEGER PRIMARY KEY AUTOINCREMENT, amount_cents INTEGER NOT NULL CHECK(amount_cents > 0), type TEXT NOT NULL CHECK(type IN ('收入', '支出')), category TEXT NOT NULL, date TEXT NOT NULL ) """)connection.commit()app=FastAPI(title="简易记账接口",version="1.0.0")@app.exception_handler(RequestValidationError)asyncdefvalidation_error(request,exception):errors=[{"field":".".join(map(str,error["loc"])),"message":error["msg"],"type":error["type"]}forerrorinexception.errors()]returnJSONResponse(status_code=422,content={"errors":errors})@app.exception_handler(sqlite3.OperationalError)asyncdefdatabase_error(request,exception):returnJSONResponse(status_code=503,content={"errors":[{"field":"database","message":"数据库暂时不可用，请稍后重试","type":"database_unavailable"}]})deffilters(type=None,category=None,dates=(None,None)):clauses,values=[],[]forcolumn,operator,valuein[("type","=",type),("category","=",category),("date",">=",dates[0]),("date","<=",dates[1])]:ifvalueisnotNone:clauses.append(f"{column}{operator}?")values.append(value.isoformat()ifisinstance(value,date)elsevalue)return(" WHERE "+" AND ".join(clauses)ifclauseselse""),valuesdefmoney(cents):returnformat(Decimal(cents)/100,".2f")@app.post("/records",status_code=201)defadd_record(record:RecordInput):cents=int(record.amount*100)withclosing(connect())asconnection,connection:cursor=connection.execute("INSERT INTO records(amount_cents, type, category, date) VALUES (?, ?, ?, ?)",(cents,record.type,record.category,record.date.isoformat()))record_id=cursor.lastrowidreturn{"id":record_id,"amount":money(cents),"type":record.type,"category":record.category,"date":record.date.isoformat()}@app.get("/records")deflist_records(type:RecordType|None=None,category:Annotated[str|None,Query(min_length=1,max_length=50)]=None,page:Annotated[int,Query(ge=1)]=1,page_size:Annotated[int,Query(ge=1,le=100)]=10,dates=Depends(date_range),):where,values=filters(type,category,dates)withclosing(connect())asconnection:connection.execute("BEGIN")total=connection.execute("SELECT COUNT(*) FROM records"+where,values).fetchone()[0]rows=connection.execute("SELECT * FROM records"+where+" ORDER BY date DESC, id DESC LIMIT ? OFFSET ?",values+[page_size,(page-1)*page_size]).fetchall()connection.commit()items=[{"id":row["id"],"amount":money(row["amount_cents"]),"type":row["type"],"category":row["category"],"date":row["date"]}forrowinrows]return{"total":total,"page":page,"page_size":page_size,"items":items}@app.get("/summary")defsummary(dates=Depends(date_range)):where,values=filters(dates=dates)withclosing(connect())asconnection:row=connection.execute(""" SELECT COALESCE(SUM(CASE WHEN type='收入' THEN amount_cents ELSE 0 END), 0), COALESCE(SUM(CASE WHEN type='支出' THEN amount_cents ELSE 0 END), 0) FROM records """+where,values).fetchone()return{"income":money(row[0]),"expense":money(row[1]),"balance":money(row[0]-row[1])}returnapp app=create_app()新增流水使用参数化 SQL，避免将用户输入直接拼接进 SQL。查询中的列名和运算符来自固定程序列表，筛选值仍使用占位符传入。列表总数和当前页数据在同一读事务中获取，避免并发写入时两次查询读到不同快照。
每次请求建立独立 SQLite 连接，写操作在事务中提交或回滚；SQLite 对写事务进行串行协调，主键由数据库分配。WAL 模式支持读写并行，连接等待锁的超时为 10 秒。如果数据库忙碌超时等操作错误发生，接口返回结构化的 503 提示。该机制适合本实验的小规模本地服务；高写入负载仍需进一步压测。
图 1 实际源代码片段及验证过程。图片由本次工程源码和执行记录排版生成，并非终端原生截图。
3. 功能调用与运行结果
启动 Uvicorn 后，使用 HTTP 客户端向本地服务依次新增以下三条流水，再调用筛选和汇总接口。记录图片中的响应来自实际 HTTP 调用。
| 日期 | 类型 | 分类 | 金额（元） |
|---|---|---|---|
| 2026-10-01 | 收入 | 工资 | 5000.00 |
| 2026-10-02 | 支出 | 餐饮 | 35.50 |
| 2026-10-03 | 支出 | 交通 | 12.80 |
查询GET /records?type=支出&page=1&page_size=1，应得到符合条件的总记录数 2，当前页仅返回日期较新的交通支出 12.80 元。
查询GET /summary?start_date=2026-10-01&end_date=2026-10-03，实际返回：
{"income":"5000.00","expense":"48.30","balance":"4951.70"}人工核对：支出为 35.50 + 12.80 = 48.30 元，结余为 5000.00 − 48.30 = 4951.70 元，与接口结果一致。
图 2 新增、筛选分页及区间汇总的实际 HTTP 响应展示。
4. 测试用例与异常处理
测试文件test_main.py使用临时数据库，覆盖正常调用和多种边界情况。
| 测试项目 | 验证内容 | 预期结果 |
|---|---|---|
| 正常新增 | 完整合法流水 | 201，返回编号及流水字段 |
| 金额边界 | 0、负数、超过两位小数、NaN、超过位数上限 | 422，不写入数据 |
| 字段异常 | 非法类型、空白分类、缺少金额、多余字段 | 422，结构化错误 |
| 日期异常 | 不存在的日期、非约定请求体日期格式 | 422 |
| 筛选分页 | 类型、分类和日期组合筛选，翻页及超出末页 | 总数、排序和当前页结果正确 |
| 汇总 | 日期边界内的收入、支出、结余 | 与预期金额相等，区间外记录被排除 |
| 空数据 | 空列表及空区间汇总 | 空列表与 0.00 |
| 查询异常 | 第 0 页、每页超过 100 条、反向日期区间、非法日期 | 422 |
| 并发写入 | 10 个线程并发提交 100 笔 0.01 元收入 | 全部 201，100 个不同编号，总收入 1.00 元 |
| 持久化 | 重建应用并读取同一数据库 | 已写入数据仍存在 |
实际执行结果为24 passed，1 warning。警告来自测试客户端对 HTTPX 的兼容性弃用提示，不影响本次测试通过。并发用例同时检查成功状态码、记录总数、编号唯一性和汇总金额。该结果验证了本次测试范围，不能据此推断任意并发规模均无风险。
另外向正在运行的 HTTP 服务提交金额为 0 的请求，实际返回 422，响应含errors数组与body.amount字段。
图 3 pytest 实际输出与非法金额请求的实际响应展示。原始记录保存在pytest-result.txt和evidence.json中。
5. AI 协作与代码审查记录
本次协作中的用户要求是完成考题二、生成四部分 Markdown 报告、提供无注释主要代码及过程和结果图片。AI 将任务拆解为需求约定、接口实现、测试、真实 HTTP 验证和报告整理。
| 阶段 | 本次处理内容 | 验证或保留记录 |
|---|---|---|
| 需求澄清 | 明确字段名、类型取值、金额精度、日期边界及分页约束 | 本报告第 1 节 |
| 编码 | 新建 main.py 与 test_main.py | 完整源文件及 changes.diff |
| 验证 | 安装依赖，执行 pytest；启动服务并调用接口 | pytest-result.txt、evidence.json、结果记录图片 |
| 复核 | 检查参数化 SQL、整数金额、事务、连接关闭及分页快照 | 主要代码及对应测试 |
changes.diff记录本次新增源码的差异，不虚构多轮修改历史。完整对话保留在当前聊天中，本节仅是关键过程摘要。
风险审查发现：当前接口未实现登录认证和账户隔离，只适合本地实验；数据库写入受 SQLite 单写者机制限制；源码未记录完整请求体，异常响应也不暴露数据库路径和 SQL 细节。运行前应根据使用场景确定数据库备份、访问控制与部署方式。代码由 AI 辅助生成，提交前仍需由提交者阅读并确认。
四、心得体会：
本实验表明，在交给 AI 编写代码之前，先明确金额精度、日期边界、分页规则和错误格式，可以减少实现过程中反复调整。接口能够启动只是第一步，还需要通过独立的预期值检查结果是否正确。
金额处理和并发写入是本次实验的重点。使用整数分存储后，汇总结果可以精确核对；使用独立连接、事务与数据库主键后，可以减少共享连接和手动分配编号引起的问题。测试既要检查正常返回，也要检查非法数据没有进入数据库。
AI 适合辅助搭建接口、补充测试和整理文档，但业务规则与交付内容仍需要逐项核对。后续可以在现有功能上增加用户认证、账户隔离和数据库备份，并在更高并发下继续验证服务行为。
