PonyORM接口适配难题：遵循SOLID原则规避@db_session重复使用

我完全懂你的痛点——PonyORM在小项目里用着特别顺手，但项目一变大，到处都是@db_session装饰器，不仅代码冗余，还容易忘加导致报错。之前我也尝试过给它做SOLID适配，确实踩了不少坑，分享几个我摸索出来的解决方案吧：

1. 用上下文管理器封装会话，告别重复装饰器 #

与其每个函数都手动加装饰器，不如把会话管理逻辑封装到你的CustomQuery里，让所有查询操作自动带上会话：

from pony.orm import db_session

class CustomQuery:
    def __init__(self, query):
        self._query = query

    def __getattr__(self, attr):
        # 动态代理Pony查询对象的属性和方法
        target_attr = getattr(self._query, attr)
        # 如果是可调用方法，自动包装db_session上下文
        if callable(target_attr):
            def wrapped_method(*args, **kwargs):
                with db_session():
                    return target_attr(*args, **kwargs)
            return wrapped_method
        return target_attr

    # 也可以单独重写常用方法，做更精细的控制（比如处理返回值）
    def count(self):
        with db_session():
            return self._query.count()

    def get(self, entity_id):
        with db_session():
            return self._query.get(id=entity_id)

这样上层业务代码调用CustomQuery的任何方法时，都会自动进入数据库会话，再也不用手动加@db_session了。

2. 基于SOLID原则的接口适配方案 #

你提到想遵循SOLID原则做适配，核心是抽象出业务需要的查询接口，让Pony的实现依赖这个接口，而不是业务代码直接依赖PonyORM。可以用抽象基类定义接口：

from abc import ABC, abstractmethod

class AbstractDataQuery(ABC):
    """定义业务层需要的查询接口"""
    @abstractmethod
    def count(self) -> int:
        pass

    @abstractmethod
    def get_by_id(self, entity_id) -> object:
        pass

    @abstractmethod
    def filter_by(self, **kwargs):
        pass

# PonyORM的适配器实现
class PonyQueryAdapter(AbstractDataQuery):
    def __init__(self, pony_query):
        self._pony_query = pony_query

    def count(self) -> int:
        with db_session():
            return self._pony_query.count()

    def get_by_id(self, entity_id) -> object:
        with db_session():
            return self._pony_query.get(id=entity_id)

    def filter_by(self, **kwargs):
        # 过滤操作不需要立即执行会话，返回新的适配器对象即可
        filtered_query = self._pony_query.filter(**kwargs)
        return PonyQueryAdapter(filtered_query)

现在业务代码只需要依赖AbstractDataQuery接口，完全不用关心底层是PonyORM还是其他ORM，符合依赖倒置原则。后续如果要替换ORM，只需要新增一个适配器实现即可。

3. 额外小技巧：全局会话管理（谨慎使用） #

如果你的项目是单线程的（比如小型Web服务的同步模式），可以考虑在项目初始化时创建一个全局会话，但这种方式要注意线程安全问题，不推荐在多线程/异步项目中使用：

from pony.orm import db_session

# 在项目启动时创建全局会话
db_session.__enter__()

# 业务代码里就不用加装饰器或上下文了
def get_user_count():
    return User.select().count()

不过这个方法风险较高，一旦处理不当容易引发会话泄漏，还是更推荐前面的适配器方案。

内容的提问来源于stack exchange，提问作者Andrew Graham-Yooll