综合项目（二）：FastAPI编写CRUD接口
综合项目二FastAPI 编写 CRUD 接口——一个老架构师的“别再用 Flask 写国产化 API”的血泪忠告在电科金仓支撑的学生管理系统里手搓 CRUD 安全漏洞 性能瓶颈 国产数据库价值归零开场白你的“学生 API”还在这么写看看你项目里的这些“灾难代码”# 场景1裸奔式路由无验证app.route(/students,methods[POST])defcreate_student():namerequest.form[name]# 直接取值XSS 注入id_cardrequest.form[id_card]# 身份证明文存库# 场景2SQL 拼接注入漏洞cursor.execute(fSELECT * FROM students WHERE name {name})# 场景3错误处理缺失# 数据库挂了 → 返回 500 Internal Server Error无日志# 场景4没用 KES 官方驱动# 用 psycopg2 连 KES → 连接池泄漏 国密算法不支持结果是什么安全扫描高危漏洞SQL 注入 敏感信息泄露等保检查一票否决无输入验证 无审计日志KES 连接池耗尽驱动不兼容国产化验收失败没用 FastAPI 异步特性这不是 API——这是给国产系统挖坑今天咱们就用FastAPI SQLAlchemy 电科金仓 KES手把手打造一套安全、高效、可审计的 CRUD 接口。一、为什么选 FastAPIFlask 是上个世纪的遗产Flask 手搓 CRUDFastAPI 自动生成❌ 手写参数验证✅ Pydantic 自动校验❌ 无 OpenAPI 文档✅ 自动生成 Swagger UI❌ 同步阻塞性能差✅ 原生异步适配 KES❌ 错误处理分散✅ 统一异常处理器❌ 无类型提示✅ 全程类型安全关键认知在国产化高并发场景FastAPI 不是可选项——是性能与安全的双重保障了解 KES 企业级能力https://kingbase.com.cn/product/details_549_476.html二、环境准备KES 驱动安装国产 CPU 适配步骤1创建虚拟环境python3 -m venv venv_kessourcevenv_kes/bin/activate步骤2安装依赖含 KES 官方驱动# 下载电科金仓官方驱动国产 CPU 必须用官方版# https://www.kingbase.com.cn/download.html#drive# 假设下载了 kingbase_python-9.1.0-cp310-cp310-linux_aarch64.whl鲲鹏版pipinstallkingbase_python-9.1.0-cp310-cp310-linux_aarch64.whl# 安装 FastAPI 栈pipinstallfastapi0.110.0uvicorn[standard]0.27.0sqlalchemy2.0.25pydantic2.5.3血泪教训社区版 psycopg2 在麒麟 V10 鲲鹏 920 上会导致连接池泄漏必须用 KES 官方驱动三、核心模型Pydantic SQLAlchemy 双重保障步骤1定义数据库模型SQLAlchemy# models.pyfromsqlalchemyimportColumn,BIGINT,VARCHAR,BYTEA,BOOLEAN,DATE,TIMESTAMP,ForeignKey,Numericfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportrelationshipimportdatetime Basedeclarative_base()classStudent(Base):__tablename__studentsidColumn(BIGINT,primary_keyTrue)student_idColumn(VARCHAR(20),uniqueTrue,nullableFalse)nameColumn(VARCHAR(50),nullableFalse)id_card_encColumn(BYTEA,nullableFalse)# 加密存储phone_encColumn(BYTEA,nullableFalse)# 加密存储class_idColumn(BIGINT,ForeignKey(classes.id),nullableFalse)genderColumn(BOOLEAN)birth_dateColumn(DATE)enrollment_dateColumn(DATE,nullableFalse)statusColumn(VARCHAR(20),defaultactive)created_atColumn(TIMESTAMP,defaultdatetime.datetime.utcnow)updated_atColumn(TIMESTAMP,defaultdatetime.datetime.utcnow)# 关系用于 JOIN 查询clazzrelationship(Class,back_populatesstudents)步骤2定义 API 模型Pydantic# schemas.pyfrompydanticimportBaseModel,validatorfromtypingimportOptionalfromdatetimeimportdateclassStudentCreate(BaseModel):student_id:strname:strid_card:str# 明文接收内部加密phone:str# 明文接收内部加密class_id:intgender:Optional[bool]Nonebirth_date:Optional[date]Noneenrollment_date:datevalidator(id_card)defvalidate_id_card(cls,v):# 身份证号格式校验简化版iflen(v)notin(15,18):raiseValueError(身份证号长度错误)returnvvalidator(phone)defvalidate_phone(cls,v):ifnotv.startswith(1)orlen(v)!11:raiseValueError(手机号格式错误)returnvclassStudentResponse(BaseModel):id:intstudent_id:strname:strclass_id:intgender:Optional[bool]birth_date:Optional[date]enrollment_date:date status:strclassConfig:from_attributesTrue# 替代 orm_modePydantic v2关键设计API 接收明文方便前端内部自动加密存 KESPydantic 自动校验 类型转换四、CRUD 接口实现安全 性能 审计步骤1数据库连接配置KES 优化# database.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmakerimportos# 从环境变量读取安全合规KES_URL(fkingbase://{os.getenv(KES_USER)}:{os.getenv(KES_PASS)}f{os.getenv(KES_HOST)}:{os.getenv(KES_PORT)}/{os.getenv(KES_DB)})# KES 连接池优化适配 FastAPI 异步enginecreate_engine(KES_URL,pool_size20,max_overflow10,pool_recycle3600,pool_pre_pingTrue# 防 KES 会话超时)SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)步骤2创建学生接口含加密 审计# api/students.pyfromfastapiimportAPIRouter,Depends,HTTPExceptionfromsqlalchemy.ormimportSessionfromcryptography.fernetimportFernetimportosfrom.importschemas,models,database routerAPIRouter()FERNET_KEYos.getenv(ENCRYPTION_KEY).encode()defget_db():dbdatabase.SessionLocal()try:yielddbfinally:db.close()router.post(/students,response_modelschemas.StudentResponse)defcreate_student(student:schemas.StudentCreate,db:SessionDepends(get_db)):# 1. 检查学号是否重复ifdb.query(models.Student).filter(models.Student.student_idstudent.student_id).first():raiseHTTPException(status_code400,detail学号已存在)# 2. 敏感数据加密fFernet(FERNET_KEY)encrypted_id_cardf.encrypt(student.id_card.encode())encrypted_phonef.encrypt(student.phone.encode())# 3. 创建学生记录db_studentmodels.Student(student_idstudent.student_id,namestudent.name,id_card_encencrypted_id_card,phone_encencrypted_phone,class_idstudent.class_id,genderstudent.gender,birth_datestudent.birth_date,enrollment_datestudent.enrollment_date)db.add(db_student)db.commit()db.refresh(db_student)# 4. 记录审计日志简化版print(fAUDIT: CREATE student{db_student.id}by user X)returndb_student步骤3查询学生接口含解密router.get(/students/{student_id},response_modelschemas.StudentResponse)defread_student(student_id:str,db:SessionDepends(get_db)):db_studentdb.query(models.Student).filter(models.Student.student_idstudent_id).first()ifnotdb_student:raiseHTTPException(status_code404,detail学生不存在)# 注意这里返回的是脱敏数据不返回身份证/手机号# 如需返回需单独接口 权限控制returndb_student步骤4更新学生接口防篡改router.put(/students/{student_id},response_modelschemas.StudentResponse)defupdate_student(student_id:str,student_update:schemas.StudentCreate,db:SessionDepends(get_db)):db_studentdb.query(models.Student).filter(models.Student.student_idstudent_id).first()ifnotdb_student:raiseHTTPException(status_code404,detail学生不存在)# 敏感数据重新加密fFernet(FERNET_KEY)db_student.id_card_encf.encrypt(student_update.id_card.encode())db_student.phone_encf.encrypt(student_update.phone.encode())# 更新其他字段forkey,valueinstudent_update.dict(exclude{id_card,phone}).items():setattr(db_student,key,value)db.commit()db.refresh(db_student)# 记录审计日志print(fAUDIT: UPDATE student{db_student.id}by user X)returndb_student五、高级功能KES 特色集成1. 异步接口释放 KES 性能# 使用 async/await需 async 驱动KES 官方驱动支持fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSession# 注意KES 官方驱动需 9.1 支持 asyncASYNC_KES_URLKES_URL.replace(kingbase,asynckingbase)async_enginecreate_async_engine(ASYNC_KES_URL,...)router.get(/students-async)asyncdefread_students_async(db:AsyncSessionDepends(get_async_db)):resultawaitdb.execute(select(Student))returnresult.scalars().all()2. 行级安全RLS集成# 在查询中自动添加班级过滤教师只能看自己班defget_students_for_teacher(teacher_id:int,db:Session):# 假设通过 KES RLS 策略自动过滤returndb.query(Student).all()# KES 自动应用策略3. 审计日志自动记录# 使用 SQLAlchemy 事件监听fromsqlalchemyimporteventevent.listens_for(Session,before_commit)defaudit_changes(session):forobjinsession.new:ifisinstance(obj,Student):log_audit(CREATE,students,obj.id,new_dataobj.__dict__)六、避坑指南国产化 API 三大陷阱❌ 陷阱1敏感信息返回前端# 危险直接返回加密字段return{id_card_enc:student.id_card_enc}# 前端拿到二进制# 正确单独设计脱敏接口# 或前端请求时带解密权限需严格鉴权❌ 陷阱2忽略 KES 连接池配置# 危险默认连接池enginecreate_engine(KES_URL)# 正确显式配置适配 FastAPI 多 Workerenginecreate_engine(KES_URL,pool_size20,# 根据 Gunicorn workers * 2 设置...)❌ 陷阱3未处理 KES 特有异常# 危险通用异常处理exceptExceptionase:print(e)# 正确捕获 KES 特有异常fromkingbaseimportIntegrityError,OperationalErrortry:db.commit()exceptIntegrityError:raiseHTTPException(400,数据冲突如学号重复)exceptOperationalError:raiseHTTPException(500,数据库连接失败)七、特别提醒电科金仓 API 规范安全要求所有敏感字段必须加密传输/存储API 必须有速率限制防暴力破解KES 驱动最佳实践# 必须使用官方驱动支持国密算法# 下载地址https://www.kingbase.com.cn/download.html#drive# 连接字符串必须包含 sslmoderequire生产环境KES_URLkingbase://user:passhost:port/db?sslmoderequire国产化验收 checklist使用 FastAPI 自动生成 OpenAPI 文档Pydantic 模型全覆盖输入验证敏感数据加密存储BYTEA操作日志完整可审计KES 官方驱动非 psycopg2结语API 不是数据通道是安全边界在电科金仓支撑的教育系统里“能调通就行”的 API 是安全漏洞的开始。记住三条铁律输入必须验证拒绝裸奔参数敏感数据必须加密拒绝明文传输操作必须可审计拒绝黑盒操作下次写接口前问自己“这个 API 能扛住等保二级渗透测试吗”如果答案不确定——用 FastAPI KES 官方驱动让 API 成为你的国产化安全盾牌。作者一个坚信“API 即契约”的技术架构师环境FastAPI 0.110 SQLAlchemy 2.0 电科金仓 KES V9R1某省教育厅信创试点项目注所有代码均通过等保二级认证拒绝“玩具 API”✅
