如何将 nameko 依赖项传递给 SqlAlchemy 事件处理程序？

Question

我正在编写一个小型 RPC 服务，它允许使用 nameko 和 sqlalchemy 对数据库进行远程 CRUD。对于某些 methods/properties/event 处理程序，我的模型需要使用依赖项来获取一些数据。我希望它的工作方式是，每当我在模型实例的生命周期内第一次调用其中一个方法时，数据就会被获取并缓存在模型上。之后，需要外部数据的方法将只使用缓存版本。

我在设计这个时遇到了一些问题。将依赖项作为函数参数传递仅适用于模型方法。属性不允许传递参数，而且我不控制 SQL Alchemy 的事件处理程序的调用方式，因此我不能在那里注入任何依赖项。我发现使这项工作起作用的唯一方法是在其生命周期的早期将依赖项绑定到模型实例，但我觉得这违背了 DI 模式。

model.py

from uuid import uuid4
import sqlalchemy as sa
from sqlalchemy import event
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()
metadata = Base.metadata


class Foo(Base):
    __tablename__ = 'foo'

    id = sa.Column(postgresql.UUID(as_uuid=True), primary_key=True, default=uuid4)
    remote_id = sa.Column(sa.Integer, nullable=False, unique=True)
    bar = sa.Column(sa.String, nullable=True)

    def __init__(self, *args, **kwargs):
        super(Foo, self).__init__(*args, **kwargs)
        self._remote_data = None

    @property
    def remote_service(self):
        # somehow return dependency
        pass

    @property
    def remote_data(self):
        if self._remote_data is None:
            self._remote_data = self.remote_service.get(id=self.remote_id)
        return self._remote_data

    @property
    def baz(self):
        return self.bar + self.remote_data.baz


def do_before_insert(mapper, connection, foo):
    # do something depending on value in foo.remote_data
    pass


event.listen(Foo, 'before_insert', do_before_insert)

service.py

from nameko.extensions import DependencyProvider
from nameko.rpc import rpc
from nameko_sqlalchemy import Database

from .model import Base, Foo


class RemoteDataService(object):
    def get(self, remote_id):
        pass

class RemoteDataServiceProvider(DependencyProvider):
    def get_dependency(self, worker_ctx):
        return RemoteDataService()

class FooRPC:
    name = "foo_rpc"
    db = Database(Base)

    @rpc
    def get_foo(self, foo_id):
        with self.db.get_session() as session:
            foo = session.query(Foo).get(foo_id)
        return foo

    @rpc
    def create_foo(self, remote_id, bar=None):
        with self.db.get_session() as session:
            foo = Foo(remote_id=remote_id, bar=bar)
            session.add(foo)
            session.commit()
        return foo

我的一个解决方案涉及在会话创建时将依赖项提供程序返回的 RemoteDataService 实例绑定到自定义数据库会话，然后使模型的依赖项属性看起来像这样：

from sqlalchemy.orm import object_session
...
    @property
    def remote_service(self):
        return object_session(self).get_remote_service()

这解决了我的问题，但它似乎不太符合 DI。此外，它仅适用于已经绑定到会话的模型，但在我的具体情况下，我可以接受。话虽这么说，如果有更好的解决方案，我会采取更好的解决方案。

在 DI/nameko/sqla 领域，我试图做的事情本身就是错误的吗？模型不应该直接处理依赖关系吗？无论哪种方式，人们如何协调使用 SQLalchemy 的事件处理程序（一种典型情况，您几乎无法控制函数调用）、DI 以及在所述处理程序中使用依赖项的需要？

Answer 1

有趣的问题。不确定我的答案是否是您所需要的。但也许这会有用。无论如何，你可以看到一个不同的方法（总比没有好）。

模型和依赖项。

不确定 models 是否适合调用某些服务或依赖项。我认为这是存储基于模型结构的微小逻辑的好地方。就是这样。类似于：

@property
def full_price(self):
    return self.coefficient * self.price

@property
def full_address(self):
    return ' '.join([self.country, self.city, self.street])

事件。 before_insert、before_update 等

与模型相同。是update/set一些模型领域的好地方。类似于：

foo.counter += 1 或 foo.updated_time = datetime.utcnow()。但不是处理一些 'remote data' 的好地方。

最好将所有其他逻辑（例如，从任何服务保存/缓存/检索数据，我的意思是 self.remote_service.get (...) 等）存储在另一层。

我试着用 small example 来解释我的意思。

Note! I didn't work with nameko and I don't know best practice etc. So this is just a vision.

查看回购中的评论。希望这会有所帮助。

如何将 nameko 依赖项传递给 SqlAlchemy 事件处理程序？

How can I pass a nameko dependency to an SqlAlchemy event handler?

python

design-patterns

dependency-injection

sqlalchemy

nameko