
一需求和标准需求实现一个简易记账 HTTP 服务数据用内存或 SQLite 存储均可。接口要求1. POST /records 新增一条流水字段金额、类型[收入/支出]、分类、日期需校验金额0、类型合法、日期格式2. GET /records 支持按类型、分类、日期区间筛选并分页返回3. GET /summary 返回指定日期区间内的收入合计、支出合计与结余4. 非法参数返回 4xx 及结构化错误信息。验收标准接口可启动并正常调用校验用例返回预期状态码用 pytest 覆盖核心用例并发写入不产生数据错误能说明如何保证即可。二实验目的本次实验基于 FastAPI 框架实现一套简易记账后端接口不开发前端页面只提供 API 服务。实验目标1.掌握 FastAPI 框架基础使用实现 POST、GET 接口使用框架自动生成 Swagger‑UI 在线接口文档。2.理解 Pydantic 模型的数据校验机制对请求参数做合法性校验认识 422 参数校验异常。3.使用 SQLite 轻量文件数据库完成数据持久化实现数据新增、分页查询、区间统计功能。4.学会使用TestClient结合 pytest 编写后端接口单元测试自动化验证接口功能。5.熟悉后端接口完整开发流程定义模型 → 编写路由 → 接口调试 → 单元测试回归验证。业务需求实现记账流水管理可以录入收支记录、分页查询流水、按时间区间统计收支总额。三项目目录结构四接口功能需求1.新增记账记录 POST /recordsamount金额必须大于 0type类型只能为income收入 / expend支出category收支分类record_date日期格式 YYYY‑MM‑DD参数非法自动返回 422 状态码2.分页查询流水 GET /records请求参数page 页码、page_size 每页条数返回分页后的记账流水列表3.时间区间收支统计 GET /summary请求参数start_date、end_date返回时间段内总收入、总支出统计结果访问 http://127.0.0.1:8000/docs 自动生成交互式接口文档。五核心代码实现from pydantic import BaseModel, Field, field_validator from datetime import date from typing import Optional, List class RecordCreate(BaseModel): amount: float Field(description流水金额必须大于0) type: str Field(description流水类型收入/支出) category: str Field(description流水分类如餐饮、工资、交通) record_date: str Field(description流水日期格式YYYY‑MM‑DD) field_validator(amount) def check_amount(cls, v): if v 0: raise ValueError(金额必须大于0) return v field_validator(type) def check_type(cls, v): if v not in (收入, 支出): raise ValueError(类型仅支持收入、支出) return v field_validator(record_date) def check_date(cls, v): try: date.fromisoformat(v) except ValueError: raise ValueError(日期格式错误请使用 YYYY‑MM‑DD) return v class RecordInfo(BaseModel): id: int amount: float type: str category: str record_date: str class Config: from_attributes True class RecordPageResponse(BaseModel): total: int page: int page_size: int total_page: int data: List[RecordInfo] class SummaryResponse(BaseModel): start_date: str end_date: str income_total: float expend_total: float balance: float class ErrorResponse(BaseModel): code: int msg: strmain.pyort FastAPI, Query, HTTPException from fastapi.responses import JSONResponse from typing import Optional from database import init_db, get_db from schemas import RecordCreate, RecordPageResponse, SummaryResponse, ErrorResponse, RecordInfo app FastAPI(title简易记账接口服务, version1.0) init_db() app.exception_handler(HTTPException) async def http_exception_handler(request, exc): return JSONResponse( status_codeexc.status_code, contentErrorResponse(codeexc.status_code, msgexc.detail).model_dump() ) app.post(/records, summary新增记账流水) def create_record(record: RecordCreate): try: with get_db() as conn: cursor conn.execute( INSERT INTO records (amount, type, category, record_date) VALUES (?, ?, ?, ?), (record.amount, record.type, record.category, record.record_date) ) new_id cursor.lastrowid row conn.execute(SELECT * FROM records WHERE id?, (new_id,)).fetchone() return RecordInfo( idrow[id], amountrow[amount], typerow[type], categoryrow[category], record_daterow[record_date] ) except Exception as e: raise HTTPException(status_code422, detailf新增流水失败{str(e)}) app.get(/records, summary筛选分页查询流水, response_modelRecordPageResponse) def get_records( type: Optional[str] Query(None, description流水类型收入/支出), category: Optional[str] Query(None, description流水分类), start_date: Optional[str] Query(None, description起始日期 YYYY‑MM‑DD), end_date: Optional[str] Query(None, description结束日期 YYYY‑MM‑DD), page: int Query(1, ge1, description页码), page_size: int Query(10, ge1, le100, description每页条数) ): sql SELECT * FROM records WHERE 11 params [] if type: if type not in (收入, 支出): raise HTTPException(status_code400, detail类型仅支持收入、支出) sql AND type? params.append(type) if category: sql AND category? params.append(category) if start_date: sql AND record_date ? params.append(start_date) if end_date: sql AND record_date ? params.append(end_date) count_sql sql.replace(SELECT *, SELECT COUNT(*)) with get_db() as conn: total conn.execute(count_sql, params).fetchone()[0] offset (page - 1) * page_size sql ORDER BY record_date DESC LIMIT ? OFFSET ? params.extend([page_size, offset]) rows conn.execute(sql, params).fetchall() data [ RecordInfo(idr[id], amountr[amount], typer[type],categoryr[category], record_dater[record_date]) for r in rows ] return RecordPageResponse(totaltotal,pagepage,page_sizepage_size,total_page(total page_size -1)//page_size,datadata) app.get(/summary, summary区间收支汇总, response_modelSummaryResponse) def get_summary( start_date: str Query(..., description起始日期 YYYY‑MM‑DD), end_date: str Query(..., description结束日期 YYYY‑MM‑DD) ): with get_db() as conn: income conn.execute( SELECT SUM(amount) FROM records WHERE type收入 AND record_date BETWEEN ? AND ?, (start_date, end_date) ).fetchone()[0] or 0.0 expend conn.execute( SELECT SUM(amount) FROM records WHERE type支出 AND record_date BETWEEN ? AND ?, (start_date, end_date) ).fetchone()[0] or 0.0 return SummaryResponse( start_datestart_date, end_dateend_date, income_totalround(income,2), expend_totalround(expend,2), balanceround(income-expend,2) ) if __name__ __main__: import uvicorn uvicorn.run(appmain:app, host0.0.0.0, port8000, reloadTrue)test_main.pyfrom fastapi.testclient import TestClient from main import app import os client TestClient(app) def setup_module(): if os.path.exists(account.db): os.remove(account.db) from database import init_db init_db() def test_create_income_record(): res client.post(/records, json{ amount: 3000, type: income, category: 工资, record_date: 2026-09-01 }) assert res.status_code 200 def test_create_expend_record(): res client.post(/records, json{ amount: 45, type: expend, category: 餐饮, record_date: 2026-09-02 }) assert res.status_code 200 def test_bad_amount_negative(): res client.post(/records, json{ amount: -100, type: income, category: 兼职, record_date: 2026-09-03 }) assert res.status_code 422 def test_bad_type(): res client.post(/records, json{ amount: 200, type: 随便, category: 购物, record_date: 2026-09-03 }) assert res.status_code 422 def test_bad_date_format(): res client.post(/records, json{ amount: 300, type: expend, category: 打车, record_date: 2026/09/03 }) assert res.status_code 422 def test_get_records_page(): res client.get(/records?page1page_size5) assert res.status_code 200 assert data in res.json() def test_summary_api(): res client.get(/summary?start_date2026-09-01end_date2026-09-30) assert res.status_code 200 assert total_income in res.json()六测试结果七实验总结1.FastAPI 开发接口非常高效自带交互式 API 文档Pydantic 可以很方便完成请求参数校验拦截非法请求。2.完整后端开发流程定义数据模型 → 编写路由接口 → 网页调试接口 → 单元测试回归验证。3.单元测试非常重要修改业务代码后一键跑全部用例避免手忙脚乱的人工重复测试。4.422 状态码是 Pydantic 校验失败的标志遇到 422 优先检查请求参数类型、格式、字段名。