SQLAlchemy使用：应做自定义封装还是直接在代码各处调用？

核心结论 #

• 行业不存在"必须全封装SQLAlchemy"或"必须全量调用原生接口"的统一惯例，两种模式都有成熟落地场景，唯一的判断标准是实际项目需求，切忌为了封装而封装。
• 你当前实现的通用query薄封装属于典型的无效封装：看似统一了查询入口，实际上把SQLAlchemy原生支持的联表查询、动态条件组合、关联预加载、分页、聚合等能力完全阻隔在封装层外。后续业务需要用到这些能力时，你要么给封装方法加一堆和原生接口一一对应的透传参数，要么绕开封装层直接写原生逻辑，最终代码里两套查询逻辑并存，反而提升维护成本。

行业通用的合理实践 #

• 基础能力直接调用原生接口即可：SQLAlchemy本身已经对数据库连接管理、会话生命周期、查询构造做了足够成熟的抽象，小型应用中直接通过上下文管理器管理Session、调用原生查询接口完全是规范做法，不存在"随处调用Session就是代码混乱"的说法。
• 仅封装高频复用的固定业务逻辑：觉得单行查询重复编写繁琐是正常的，但不要做支持传入任意SQL的通用查询壳子，而是针对具体的高频查询场景封装独立方法。比如你经常按device_id查询设备表，就单独实现get_devices_by_device_id方法；仅出现一两次的复杂查询，直接在业务代码中写原生SQLAlchemy逻辑即可，不需要强行封装。
• 杜绝直接拼接SQL字符串的写法：你示例中直接传入"where device_id="FOOBAR""的写法存在明确的SQL注入风险，即使用text()构造原生SQL，也要通过参数占位符传值，绝对不能把外部输入直接拼接到SQL语句中。

轻量实现参考 #

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker, scoped_session
from your_models import MyOrmTableClass

# 仅做最基础的引擎和会话初始化，不需要套多余的自定义DB类
engine = create_engine("sqlite:///path_to_db.sqlite")
db_session = scoped_session(sessionmaker(bind=engine, autocommit=False, autoflush=False))

# 只封装高频复用的固定查询逻辑，不做通用透传壳子
def get_devices_by_id(device_id: str):
    with db_session() as sesh:
        return sesh.scalars(
            select(MyOrmTableClass)
            .where(MyOrmTableClass.device_id == device_id)
        ).all()

# 业务代码直接调用即可，复杂查询直接用原生接口实现
results = get_devices_by_id("FOOBAR")

实用经验提示 #

• 如果觉得SQLAlchemy语法生硬，先确认是否还在使用旧版本的session.query()写法。1.4版本后推出的select()风格API和原生SQL语义高度对齐，熟练使用后基本不会有违和感。
• 基于SQLite的小型应用不要硬套Repository、DDD之类的重型分层架构，这类项目中直接编写查询逻辑的维护成本，远低于维护一堆仅做参数透传的薄封装层的成本。
• 只有当你有明确的跨数据库兼容需求、统一操作审计/日志需求、多租户数据自动过滤需求时，再考虑做全局层面的SQLAlchemy封装，否则直接使用原生能力的开发效率最高。

内容的提问来源于stack exchange，提问作者John Brodie