FastAPI+SQLAlchemy 异步 CRUD 完整示例-尧图网

统一采用session: AsyncSession Depends(get_session)依赖注入方式不使用中间件request.state.session事务使用async with session.begin()自动 com依赖准备前面已写from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, delete, update from pydantic import BaseModel from typing import List from models import AsyncSessionFactory, User app FastAPI() # 依赖函数获取数据库Session async def get_session(): session AsyncSessionFactory() try: yield session finally: await session.close() # 响应Schema class UserRespSchema(BaseModel): id: int email: str username: str class Config: orm_mode True # 创建用户请求Schema class UserCreateReqSchema(BaseModel): email: str username: str password: str1、新增数据 createapp.post(/user/add, response_modelUserRespSchema) async def add_user( req: UserCreateReqSchema, session: AsyncSession Depends(get_session) ): try: async with session.begin(): user User(usernamereq.username, emailreq.email, passwordreq.password) session.add(user) return user except Exception: raise HTTPException(status_code400, detail用户名或邮箱已经存在)2、删除数据 deleteapp.delete(/user/delete/{user_id}) async def delete_user( user_id: int, session: AsyncSession Depends(get_session) ): async with session.begin(): await session.execute( delete(User).where(User.id user_id) ) return {message: 删除成功}3、查询数据 select查询单条app.get(/user/select/{user_id}, response_modelUserRespSchema) async def select_user_by_id( user_id: int, session: AsyncSession Depends(get_session) ): async with session.begin(): query await session.execute( select(User.id, User.email, User.username).where(User.id user_id) ) result query.one()._asdict() return result查询多条模糊搜索、分页排序app.get(/user/list, response_modelList[UserRespSchema]) async def select_user_list( q: str | None None, session: AsyncSession Depends(get_session) ): from sqlalchemy import or_ stmt select(User.id, User.username, User.email) if q: stmt stmt.where( or_( User.email.contains(q), User.username.contains(q) ) ) # 分页、排序 stmt stmt.limit(2).offset(0).order_by(User.id.desc()) query await session.execute(stmt) result [row._asdict() for row in query] return result注意session.query()只适用于同步 SQLAlchemyselect()。4、修改数据 update方式 1先查询对象修改对象属性执行 2 次 SQL可以直接返回修改后对象app.put(/user/update/{user_id}, response_modelUserRespSchema) async def update_user_obj( user_id: int, user_data: UserCreateReqSchema, session: AsyncSession Depends(get_session) ): async with session.begin(): query await session.execute(select(User).where(User.id user_id)) user query.scalars().first() user.email user_data.email user.username user_data.username return user方式 2直接 update 语句只执行 1 次 SQL不会刷新 ORM 对象app.put(/user/update_direct/{user_id}) async def update_user_direct( user_id: int, user_data: UserCreateReqSchema, session: AsyncSession Depends(get_session) ): async with session.begin(): await session.execute( update(User) .where(User.id user_id) .values(**user_data.dict()) ) return {message: 数据修改成功}

FastAPI+SQLAlchemy 异步 CRUD 完整示例

更多相关文章

SCDLG：生成式抄袭检测与候选源重排序新方法

DeepSeek与Kimi接入指南：从API调用到IDE插件配置全解析

从SEIR到SEIDR：Python实现传染病建模与外部输入分析

ARM Cortex-M3 DesignStart包评估与SoC集成避坑指南

SpringBoot+Vue+MyBatis+MySQL企业人事系统架构实战解析

Kappa架构：基于事件时间的实时数仓统一处理范式

免费磁盘分区安全指南：备份、GPT/MBR识别与避坑实操

大模型微调实战指南：LoRA参数高效微调的原理、策略与避坑技巧

从AI生成到手工质感：ArtCraft创作工作流全解析

Python调用Gemini Structured Outputs实现工单路由门禁

Spring Boot超市进销存系统毕设实战：从需求拆解到答辩通关

Flutter StatefulWidget 生命周期核心解析

语义分割与医学图像分割论文阅读指南：从FCN到Transformer的演进路线

PlanarSLAM：面向曼哈顿世界的点线面联合SLAM系统

TDH Workflow调度实战：从DAG依赖驱动到任务编排与故障排查

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

还想了解更多？直接咨询顾问