sqlalchemy

SQLAlchemy Core 知识

深度知识：使用 mcp__documentation__fetch_docs，并将技术指定为：sqlalchemy，以获取完整文档。

何时不应使用此技能

• TypeScript/Node.js 项目：使用 prisma、drizzle 或 typeorm 技能
• Django 应用：使用 Django ORM 文档（此处不涵盖）
• 原始 SQL 查询：使用 database-query MCP 服务直接执行 SQL
• NoSQL 数据库：使用 mongodb 技能执行 MongoDB 操作
• 特定于 FastAPI 的内容：集成模式可能需要 fastapi-expert
• 数据库设计：咨询 sql-expert 或 architect-expert 了解架构设计
• 其他 Python ORM：此技能不支持 Peewee、Pony ORM 和 Tortoise

模型定义

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(255), nullable=False)
    content = Column(String)
    author_id = Column(Integer, ForeignKey('users.id'))

    author = relationship('User', back_populates='posts')

会话操作

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)

# Create
with Session() as session:
    user = User(name='John', email='[email protected]')
    session.add(user)
    session.commit()

# Read
with Session() as session:
    users = session.query(User).all()
    user = session.query(User).filter_by(id=1).first()
    active_users = session.query(User).filter(User.is_active == True).all()

# Update
with Session() as session:
    user = session.query(User).filter_by(id=1).first()
    user.name = 'Jane'
    session.commit()

# Delete
with Session() as session:
    session.query(User).filter_by(id=1).delete()
    session.commit()

异步支持

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/db')
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_users():
    async with async_session() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

反模式

反模式	不良之处	更好的做法
未关闭会话	连接泄漏、连接池耗尽	使用上下文管理器（with Session()）
在 SQLAlchemy 2.0+ 中使用 session.query()	已弃用的旧版 API	使用 select() 和 session.execute()
未对关联进行预加载	N+1 查询问题	使用 selectinload() 或 joinedload()
将连接字符串硬编码	安全风险	使用环境变量
未配置连接池	连接问题、性能问题	设置 pool_size、max_overflow、pool_recycle
未理解就使用 session.flush()	部分提交、造成混淆	使用 commit() 处理事务
在生产环境中未设置 pool_pre_ping	数据库重启后连接失效	启用 pool_pre_ping=True
外键缺少索引	联接速度慢	为频繁参与联接的列添加 Index()
使用 ORM 执行批量操作	处理大型数据集时速度很慢	使用 bulk_insert_mappings() 或 Core
未处理 IntegrityError	向用户显示晦涩的错误	捕获错误并提供有意义的消息

快速故障排除

问题	可能原因	解决方案
"No module named 'sqlalchemy'"	未安装 SQLAlchemy	运行 pip install sqlalchemy
"Can't connect to server"	DATABASE_URL 错误或数据库已关闭	验证连接字符串，检查数据库状态
"Table doesn't exist"	尚未运行迁移	执行 alembic upgrade head
"DetachedInstanceError"	在会话外访问关联	使用 joinedload() 或保持会话打开
"IntegrityError: duplicate key"	违反唯一约束	检查记录是否已存在，并处理错误
"InvalidRequestError: SQL expression"	2.0 风格中缺少 select()	使用 select(Model)，而非 session.query()
查询速度慢	N+1 问题、缺少索引	添加预加载，创建索引
连接池超时	连接过多	增大 pool_size，修复连接泄漏
"Stale data"	会话缓存了旧数据	调用 session.expire_all() 或刷新对象
Alembic 冲突	存在多个迁移头	使用 alembic merge 合并分支

生产环境就绪

引擎配置

# database.py
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool
import os

DATABASE_URL = os.environ['DATABASE_URL']

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,  # Recycle connections after 30 minutes
    pool_pre_ping=True,  # Check connection validity
    echo=os.environ.get('DEBUG') == 'true',
    connect_args={
        'sslmode': 'require' if os.environ.get('ENV') == 'production' else 'prefer',
        'connect_timeout': 10,
    }
)

# Event listeners for monitoring
@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(engine, 'after_cursor_execute')
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop()
    if total > 0.5:  # Log slow queries
        logger.warning(f'Slow query ({total:.2f}s): {statement[:100]}')

SessionLocal = sessionmaker(bind=engine)
ScopedSession = scoped_session(SessionLocal)

上下文管理器模式

from contextlib import contextmanager
from typing import Generator

@contextmanager
def get_db() -> Generator[Session, None, None]:
    """Database session context manager with automatic cleanup."""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
def create_user(name: str, email: str) -> User:
    with get_db() as db:
        user = User(name=name, email=email)
        db.add(user)
        db.flush()  # Get ID without committing
        return user

异步配置

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

async_engine = create_async_engine(
    DATABASE_URL.replace('postgresql://', 'postgresql+asyncpg://'),
    pool_size=20,
    max_overflow=10,
    pool_recycle=1800,
    echo=False,
)

async_session = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

查询优化

from sqlalchemy.orm import joinedload, selectinload
from sqlalchemy import select

# Eager loading to avoid N+1
async def get_users_with_posts(db: AsyncSession):
    stmt = (
        select(User)
        .options(selectinload(User.posts))
        .where(User.is_active == True)
    )
    result = await db.execute(stmt)
    return result.scalars().all()

# Pagination
async def paginate(
    db: AsyncSession,
    page: int,
    per_page: int
) -> dict:
    offset = (page - 1) * per_page

    # Count
    count_stmt = select(func.count(User.id))
    total = await db.scalar(count_stmt)

    # Data
    stmt = (
        select(User)
        .order_by(User.created_at.desc())
        .offset(offset)
        .limit(per_page)
    )
    result = await db.execute(stmt)

    return {
        'data': result.scalars().all(),
        'total': total,
        'page': page,
        'pages': math.ceil(total / per_page),
    }

# Bulk insert
def bulk_insert_users(db: Session, users_data: list[dict]):
    db.execute(insert(User), users_data)
    db.commit()

Alembic 迁移

# alembic/env.py
from sqlalchemy import engine_from_config, pool
from alembic import context
from models import Base

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True,
        )

        with context.begin_transaction():
            context.run_migrations()

# CLI commands
# alembic revision --autogenerate -m "Add users table"
# alembic upgrade head
# alembic downgrade -1

测试

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope='session')
def engine():
    return create_engine(os.environ['TEST_DATABASE_URL'])

@pytest.fixture(scope='session')
def tables(engine):
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db(engine, tables):
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

# tests/test_user.py
def test_create_user(db):
    user = User(name='Test', email='[email protected]')
    db.add(user)
    db.flush()
    assert user.id is not None

监控指标

指标	目标
查询时间（p99）	< 100ms
连接池连接数	< max_size
慢查询	0（> 500ms）
连接错误	0

检查清单

• 已配置连接池
• 生产环境已启用 SSL
• 已启用 pool_pre_ping
• 长时间运行的应用已设置 pool_recycle
• 使用上下文管理器管理会话
• 使用预加载防止 N+1 问题
• 为列表查询添加分页
• 使用 Alembic 迁移
• 记录慢查询
• 通过回滚测试隔离性

参考文档

• 关联关系
• 异步