SQLAlchemy 2.0 异步模式实战

# SQLAlchemy 2.0 异步模式实战 ## 异步引擎配置 ```python from sqlalchemy.ext.asyncio import ( create_async_engine, async_sessionmaker, AsyncSession ) from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship from sqlalchemy import String, ForeignKey, select from datetime import datetime engine = create_async_engine( "postgresql+asyncpg://user:pass@localhost/mydb", pool_size=20, max_overflow=10, pool_pre_ping=True, echo=False ) AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) class Base(DeclarativeBase): pass ``` ## 模型定义（2.0 风格） ```python class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(100)) email: Mapped[str] = mapped_column(String(200), unique=True) created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin") class Post(Base): __tablename__ = "posts" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(200)) content: Mapped[str] = mapped_column() author_id: Mapped[int] = mapped_column(ForeignKey("users.id")) author: Mapped["User"] = relationship(back_populates="posts") ``` ## 异步 CRUD ```python class UserRepository: def __init__(self, session: AsyncSession): self.session = session async def get_by_id(self, user_id: int) -> User | None: stmt = select(User).where(User.id == user_id) result = await self.session.execute(stmt) return result.scalar_one_or_none() async def search(self, keyword: str, page: int = 1, size: int = 20): stmt = ( select(User) .where(User.name.ilike("%" + keyword + "%")) .order_by(User.created_at.desc()) .offset((page - 1) * size) .limit(size) ) result = await self.session.execute(stmt) return result.scalars().all() async def create(self, name: str, email: str) -> User: user = User(name=name, email=email) self.session.add(user) await self.session.flush() return user ``` ## FastAPI 集成 ```python from fastapi import FastAPI, Depends app = FastAPI() async def get_session(): async with AsyncSessionLocal() as session: async with session.begin(): yield session @app.get("/users/{user_id}") async def get_user(user_id: int, session: AsyncSession = Depends(get_session)): repo = UserRepository(session) user = await repo.get_by_id(user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return {"id": user.id, "name": user.name, "email": user.email} ``` SQLAlchemy 2.0 的异步模式配合 asyncpg 驱动，在高并发场景下性能表现优异，推荐新项目直接采用异步方案。