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: 数据修改成功}