Skip to main content
 首页 » 编程设计

python之SQLAlchemy ORM 重构器的替代品

2025年05月04日7dyllove98

我目前正在使用 Python (3.6) 开发一个小应用程序,能够处理股票投资组合,并对其进行一些财务分析。

这些功能只是我的投资组合和股票的 CRUD 界面(创建、读取、更新、删除)。它们可以保存在数据库中(使用 PostgreSQL 和 SQLAlchemy 作为 ORM)

问题是,我的目标是在以下之间绝对解耦:

  • “业务逻辑”(位于包 core 中),其中我有我的 Python 对象 Stock、Portfolio、回归等
  • “持久层”(位于单独的包 database 中),其中定义了所有 SQLAlchemy 元数据和与 core 包对象的映射

我想要这种分离有几个原因:首先,我希望能够独立启动 core 包(既没有安装任何数据库,也没有安装 SQLAlchemy),其次,我认为它是一个很好的做法,因为在我的 core 包中,我真的可以只专注于分析,而不是用 SQLA Column、relationships“污染”我的 Python 对象> 等

到目前为止,我使用低级 SQLAlchemy mapper 和 Table 设法做到了。 但是,只有一点我无法弄清楚。目前我的 Python Stock 对象是这样的:

class Stock: 
 
    def __init__(self, ticker: str, exchange: Exchange, name: str=None): 
        self.ticker = ticker 
        self.exchange = exchange 
        self.name = name 
        self._data = None 
 
    ... 

需要 _data 属性来存储我的财务数据,我想实例化它。

但是,正如您所知,当我从数据库加载 Stock 时(在我实际使用 database pkg 的情况下),例如通过命令如 session.query(Stock).first(),则 __init__ 函数不会被 SQLA 调用。我唯一能做的就是添加一个函数:

@orm.reconstructor 
def init_on_load(self): 
    self._data = None 

其中 orm 是 sqlalchemy.orm 并且是 唯一 在我的 core 中导入 sqlalchemy 包。我想摆脱它!

有人有想法吗?例如,也许有一种方法可以通过 mapper 从 database 包内部实例化 _data 属性,但我不知道任何。

谢谢

请您参考如下方法:

编辑:

对评论的回应,“数据映射器不是侵入性的,因此您可以在数据层和业务层中使用一个数据类”。只有在项目很小的情况下才是正确的。但是当项目变大时,底层数据存储不会以与您的业务逻辑相同的速度变化。然后你需要一些中间层,例如适配器、变压器等。

无论如何,如果你的问题不是关于解耦,只是想让它工作,那么使用@orm.reconstructor 或任何技巧。否则,尽可能使用最少的黑客攻击或解决方法。

======================================

根据我的经验,ORM类与底层OMR框架耦合,无论你选择什么框架,SQLAlchemy、MongoEngine等。

如果你想抽象数据存储,以后可以切换数据引擎。然后最好在一些普通的旧类上编写您的核心业务逻辑。然后在存储模块中,提供一个将 ORM 类转换为普通类的函数。

例如

# module core.objects 
class Stock: 
    def __init(self, name, price): 
        self.name = name 
        self.price = price 
 
# business logic 
class BusinessLogic: 
    @staticmethod 
    def logic(stocks: list[Stock]): 
        pass 

然后在存储包中

from core.objects import Stock as CoreStock 
# This is the ORM dependent class 
class Stock(Model): 
    name = columns.Text() 
    price = columns.Float() 
 
    @classmethod 
    def to_core_objects(instances): 
        return [CoreStock(instance.name, instance.price) for instance in instances] 

现在你可以轻松切换存储引擎,mysql、csv等,编写单元测试也更容易。