对比维度​

SQLAlchemy 异步 ORM 优势​

其他异步 ORM 常见短板​

生态成熟度与兼容性​

1. 基于 SQLAlchemy 多年沉淀的核心生态，支持 MySQL、PostgreSQL、SQLite 等所有主流数据库；Alchemy 同步版的大部分 API，同步项目迁移异步成本极低；数据库特性（如事务隔离级别、存储过程、触发器）。​

1. 部分轻量框架（如 Tortoise）对小众数据库支持不足；>2. Django ORM 异步版依赖 Django 生态，脱离 Django 无法独立使用。​

查询灵活性与功能完备性​

1. 支持原生 SQL 混合查询、复杂 JOIN、子查询、窗口函数等高级 SQL 特性，兼顾 ORM 便捷性与 SQL 灵活性；. 提供强大的查询构造器，支持条件组合、聚合计算、分页排序等全场景需求（如你文档中 4.5 节的各类查询示例）；、约束的精细化配置（如模型类中 mapped_column 的各类参数）。​

1. Tortoise ORM 等轻量框架查询语法简洁，但复杂查询能力有限，需手动拼接 SQL；2. Peewee-Async 功能较基础，不支持部分高级聚合查询。​

异步性能与资源管理​

1. 内置高效连接池（如文档中 pool_size、max_overflow 配置），支持连接复用与动态扩容，适配高并发场景；>2. 异步会话（AsyncSession）配合 yield 语法实现资源自动释放，避免连接泄露；生命周期管理（提交、回滚、嵌套事务）。​

1. 部分框架连接池配置简单，高并发下易出现连接耗尽；. 部分 ORM 异步事务支持不完整（如不支持嵌套事务）。​

与 FastAPI 协同性​

1. 天然适配 FastAPI 的异步特性与依赖注入系统（如文档中 get_database 依赖项，直接注入异步会话）； Pydantic 深度兼容，支持模型类与 Pydantic 模型快速转换（如 4.5.5 节 BookBase 转 Book）；API 接口文档生成，无需额外配置。​

1. 其他 ORM 需手动适配 FastAPI 依赖注入，代码冗余； 部分框架与 Pydantic 兼容性一般，需手动处理数据类型转换。​

扩展性与定制化能力​

1. 支持自定义类型映射、字段验证逻辑、查询拦截器等高级扩展； 系统监听数据库操作生命周期（如模型创建、更新前的钩子函数）；支持数据库迁移工具（Alembic），适配项目迭代中的 schema 变更。​

1. 轻量 ORM 扩展点少，定制化能力弱；工具，需手动维护 SQL 脚本。​

学习成本与社区支持​

1. 文档完善、社区活跃，问题解决方案丰富（如你参考的 CSDN 实战文章、官方文档）； API 风格一致，熟悉同步 SQLAlchemy 的开发者可快速上手；工具集成（如日志、监控、测试工具）。​

1. 部分小众框架文档简陋，社区支持有限；异步版需先掌握 Django 生态，学习成本高。​

from datetime import datetime

from sqlalchemy import DateTime, func, String, Float, select

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from fastapi import Depends, HTTPException

from pydantic import BaseModel

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# MySQL异步连接URL格式：数据库类型+异步驱动://用户名:密码@地址:端口/数据库名?编码

ASYNC_DATABASE_URL = "mysql+aiomysql://root:1234@localhost:3306/fastapi_test?charset=utf8"

async_engine = create_async_engine(

    ASYNC_DATABASE_URL,

    echo=True,        # 开启SQL日志（调试用）

    pool_size=10,     # 连接池常驻连接数

    max_overflow=20   # 高并发时额外允许的连接数

)

AsyncSessionLocal = async_sessionmaker(

    bind=async_engine,        # 绑定异步引擎

    class_=AsyncSession,      # 指定异步会话类型

    expire_on_commit=False    # 提交后会话不过期

)

async def get_database():

    async with AsyncSessionLocal() as session:

        try:

            yield session  # 注入会话给路由

            await session.commit()  # 无异常提交事务

        except Exception:

            await session.rollback()  # 异常回滚

            raise

        finally:

            await session.close()  # 关闭会话释放连接

class Base(DeclarativeBase):

    # 公共字段：创建时间、更新时间

    create_time: Mapped[datetime] = mapped_column(

        DateTime, insert_default=func.now(), default=func.now(), comment="创建时间"

    )

    update_time: Mapped[datetime] = mapped_column(

        DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间"

    )

class Book(Base):

    __tablename__ = "book"  # 数据库表名

    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍ID")  # 主键

    bookname: Mapped[str] = mapped_column(String(255), comment="书名")

    author: Mapped[str] = mapped_column(String(255), comment="作者")

    price: Mapped[float] = mapped_column(Float, comment="价格")

    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 建表函数

async def create_tables():

    async with async_engine.begin() as conn:

        await conn.run_sync(Base.metadata.create_all)  # 创建所有模型对应的表

# FastAPI启动事件：项目启动时执行建表

@app.on_event("startup")

async def startup_event():

    await create_tables()

# 1. 查询所有书籍

@app.get("/book/books")

async def get_book_list(db: AsyncSession = Depends(get_database)):

    result = await db.execute(select(Book))

    return result.scalars().all()

# 2. 查询单条书籍（第一条）

@app.get("/book/bookOne")

async def get_book_single(db: AsyncSession = Depends(get_database)):

    result = await db.execute(select(Book))

    return result.scalars().first()

# 3. 按主键查询（高效）

@app.get("/book/bookGet")

async def get_book_by_id(db: AsyncSession = Depends(get_database)):

    return await db.get(Book, 1)  # 参数：模型类 + 主键值

查询类型

核心语法示例

多条件比较

select(Book).where(Book.id==book_id, Book.bookname==book_name, Book.price<=book_price)

模糊查询

select(Book).where(Book.author.like(f"{author}%"))（% 匹配任意字符）

逻辑与

`select(Book).where(Book.author.like(f"{author}%") & (Book.price

逻辑或

`select(Book).where(Book.author.like(f"{author}%")

# 1. 价格平均值

@app.get("/book/count")

async def get_book_price_avg(db: AsyncSession = Depends(get_database)):

    result = await db.execute(select(func.avg(Book.price)))

    return {"avg_price": result.scalar()}

# 2. 价格最大值

@app.get("/book/max")

async def get_book_price_max(db: AsyncSession = Depends(get_database)):

    result = await db.execute(select(func.max(Book.price)))

    return {"max_price": result.scalar()}

# 3. 价格总和

@app.get("/book/sum")

async def get_book_price_sum(db: AsyncSession = Depends(get_database)):

    result = await db.execute(select(func.sum(Book.price)))

    return {"sum_price": result.scalar()}

@app.get("/book/get_books")

async def get_book_paginate(

    page: int = 1, page_size: int = 3, db: AsyncSession = Depends(get_database)

):

    skip = (page - 1) * page_size  # 分页核心公式：跳过条数 = (页码-1)*每页条数

    stmt = select(Book).offset(skip).limit(page_size)

    result = await db.execute(stmt)

    return result.scalars().all()

# Pydantic模型（数据校验）

class BookBase(BaseModel):

    id: int; bookname: str; author: str; price: float; publisher: str

# 新增接口

@app.post("/book/add_book")

async def add_book(book: BookBase, db: AsyncSession = Depends(get_database)):

    book_obj = Book(**book.dict())  # Pydantic转ORM对象

    db.add(book_obj)  # 添加到事务

    return book  # 依赖项自动提交事务

# Pydantic模型（更新数据校验）

class BookUpdate(BaseModel):

    bookname: str; author: str; price: float; publisher: str

# 更新接口

@app.put("/book/update_book/{book_id}")

async def update_book(id: int, book: BookUpdate, db: AsyncSession = Depends(get_database)):

    book1 = await db.get(Book, id)

    if not book1: raise HTTPException(status_code=404, detail="Book not found")

    # 重新赋值更新

    book1.bookname = book.bookname; book1.price = book.price; book1.author = book.author; book1.publisher = book.publisher

    return book1

@app.delete("/book/delete_book/{book_id}")

async def delete_book(book_id: int, db: AsyncSession = Depends(get_database)):

    db_book = await db.get(Book, book_id)

    if not db_book: raise HTTPException(status_code=404, detail="book not found")

    await db.delete(db_book)  # 删除对象

    return {"message": "book is deleted"}