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实现传染病建模与外部输入分析

更多资讯

鹤壁市电工证考试怎么准备？电子证书时代怎么不掉队

中级电工证在学校考怎么报名才能不被坑

电工证的版本怎么考，过期了怎么报名复审

读书维修电工证怎么考内部消息曝光工地太忙也能考过

宝鸡陇县电工证过期了怎么复审？全国通用证书怎么查？

中焊工电工证考试培训机构怎么选

今日更新

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本周精选

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？

相关推荐

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

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

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

更多资讯

鹤壁市电工证考试怎么准备？电子证书时代怎么不掉队

中级电工证在学校考怎么报名才能不被坑

电工证的版本怎么考，过期了怎么报名复审

读书维修电工证怎么考内部消息曝光 工地太忙也能考过

宝鸡陇县电工证过期了怎么复审？全国通用证书怎么查？

中焊工电工证考试培训机构怎么选

今日更新

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本周精选

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？

读书维修电工证怎么考内部消息曝光工地太忙也能考过