Python中分享SQLAlchemy使用经验与常见问题

剥离 Django 的 orm 是在麻烦
于是生抗 SQLAlchemy
github.com/firejoke/moko_model


Python中分享SQLAlchemy使用经验与常见问题
1 回复

SQLAlchemy的核心在于理解它的两个主要部分：Core（SQL表达式语言）和ORM（对象关系映射）。很多人一开始就扎进ORM，结果遇到各种性能问题和复杂查询不会写。我的建议是：先掌握Core，再使用ORM。

1. 连接池配置是必须的 默认的连接池可能不适合生产环境。一定要配置连接池大小和回收时间：

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600,
    pool_pre_ping=True  # 防止连接超时
)

2. 批量操作别用ORM的add_all() 这是最常见的性能陷阱。批量插入数据时，直接用Core的execute()：

# 错误做法（慢）：
for item in data_list:
    session.add(MyModel(**item))

# 正确做法（快100倍）：
from sqlalchemy import insert
stmt = insert(MyModel.__table__).values(data_list)
session.execute(stmt)

3. 复杂查询用Core写 ORM的查询接口在复杂JOIN和子查询时很笨重。学会用Core写原生SQL表达式：

from sqlalchemy import select, func, case

# 统计每个用户的订单数和总金额
stmt = select(
    User.id,
    User.name,
    func.count(Order.id).label('order_count'),
    func.sum(
        case(
            (Order.status == 'completed', Order.amount),
            else_=0
        )
    ).label('total_amount')
).join(Order, isouter=True).group_by(User.id)

# 执行查询
result = session.execute(stmt)
for row in result:
    print(f"{row.name}: {row.order_count} orders, ${row.total_amount}")

4. 事务管理要明确 别依赖自动提交，显式控制事务：

try:
    with session.begin():
        # 一系列操作
        session.add(obj1)
        session.execute(stmt)
        # 这里出错会自动回滚
except Exception as e:
    # 处理异常
    session.rollback()
    raise

5. 模型定义用TypeDecorator扩展类型 遇到数据库特有类型时，别硬编码：

from sqlalchemy import TypeDecorator, String
import json

class JSONEncodedDict(TypeDecorator):
    """JSON字典类型"""
    impl = String
    
    def process_bind_param(self, value, dialect):
        return json.dumps(value) if value else None
    
    def process_result_value(self, value, dialect):
        return json.loads(value) if value else None

# 在模型中使用
class MyModel(Base):
    __tablename__ = 'mytable'
    id = Column(Integer, primary_key=True)
    data = Column(JSONEncodedDict)  # 自动序列化/反序列化

6. 懒加载的坑 默认的关系加载是懒加载，容易产生N+1查询问题。用joinedload()或selectinload()：

from sqlalchemy.orm import joinedload

# 一次查询加载所有关联数据
users = session.query(User).options(
    joinedload(User.orders).joinedload(Order.items)
).all()

7. 调试SQL输出 开发阶段一定要看生成的SQL：

import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

8. 版本迁移用Alembic 别手动改表结构，用Alembic做迁移：

# 初始化
alembic init migrations
# 生成迁移脚本
alembic revision --autogenerate -m "add user table"
# 执行迁移
alembic upgrade head

9. 多数据库支持 一个项目连多个数据库很常见：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engines = {
    'primary': create_engine('postgresql://...'),
    'replica': create_engine('postgresql://...'),
}

SessionPrimary = sessionmaker(bind=engines['primary'])
SessionReplica = sessionmaker(bind=engines['replica'])

10. 连接字符串特殊字符 密码里有特殊字符时记得转义：

# 密码中有@符号
from urllib.parse import quote_plus
password = quote_plus('p@ssw0rd#123')
engine = create_engine(f'postgresql://user:{password}@localhost/db')

总结： 把SQLAlchemy当工具集而不是黑盒子，知道什么时候用ORM什么时候用Core。

回到顶部