Mixin Cython 类和 SqlAlchemy

Question

摘要：

我有一个代表业务单位的 cython 类。此类以纯 cython 样式声明。

在一个项目中，我需要将业务部门映射到数据库。为此，我想导入 .pxd 文件并使用 SQLAlchemy“映射”它。

Cython 定义

让我们假设类设备。该类在 .pxd 中定义了 .pyx 和类接口（因为我需要在其他模块中导入它）。

设备.pxd

cdef class Equipment:
    cdef readonly int x
    cdef readonly str y

设备.pyx

cdef class Equipment:
    def __init__(self, int x, str y):
        self.x = x
        self.y = y

我编译所有内容并获得一个设备.pyd 文件。到目前为止，还可以。 此文件包含业务逻辑模型，不得更改。

映射

然后在一个应用程序中，我导入设备.pyd 并使用 SQLAlchemy 映射它。

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from equipment import Equipment

metadata = MetaData()

# Table definition
equipment = Table(
    'equipment', metadata,
    Column('id', Integer, primary_key=True),
    Column('x', Integer),
    Column('y', String),
)

# Mapping the table definition with the class definition
mapper(Equipment, equipment)

TypeError: can't set attributes of built-in/extension type 'equipment.Equipment'

确实，SQLAlchemy 正在尝试创建 Equipment.c.x、Equipment.c.y、...这在 Cython 中是不可能的，因为它没有在 .pxd 中定义...

那么如何将 Cython 类映射到 SQLAlchemy？

不满意的解决方案

如果我在 .pyx 文件中以 python 模式定义设备类，它可以工作，因为最后，它只是 cython 类定义中的“python”对象。

设备.pyx

class Equipment:
    def __init__(self, x, y):
        self.x = x
        self.y = y

但是我失去了很多功能，这就是我需要纯 Cython 的原因。

谢谢！ :-)

-- 编辑部分--

半满意解

保留 .pyx 和 .pxd 文件。从 .pyd 继承。尝试映射。

mapping.py

​​>

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from equipment import Equipment

metadata = MetaData()

# Table definition
equipment = Table(
    'equipment', metadata,
    Column('id', Integer, primary_key=True),
    Column('x', Integer),
    Column('y', String),
)

# Inherit Equipment to a mapped class
class EquipmentMapped(Equipment):
    def __init__(self, x, y):
        super(EquipmentMapped, self).__init__(x, y)

# Mapping the table definition with the class definition
mapper(EquipmentMapped, equipment)

从映射导入 EquipmentMapped

e = EquipmentMapped(2, 3)

打印 e.x

##这是空的！

为了让它工作，我必须将每个属性定义为一个属性！

设备.pxd

cdef class Equipment:
    cdef readonly int _x
    cdef readonly str _y

设备.pyx

cdef class Equipment:
    def __init__(self, int x, str y):
        self.x = x
        self.y = y
    property x:
        def __get__(self):
            return self._x
        def __set__(self, x):
            self._x = x
    property y:
        def __get__(self):
            return self._y
        def __set__(self, y):
            self._y = y

这并不令人满意，因为 :lazy_programmer_mode on: 我在业务逻辑上要做很多更改... :lazy_programmer_mode off:

Answer 1

我认为基本问题是，当您致电 mapper 时（除其他外）

Equipment.x = ColumnProperty(...) # with some arguments
Equipment.y = ColumnProperty(...)

当 ColumnProperty 是 sqlalchemy 定义的属性时，所以当您执行 e.x = 5 时，它可以注意到该值在它周围的所有数据库相关内容中都发生了变化。

显然不能很好地与您试图用来控制存储的下面的 Cython 类配合使用。

就个人而言，我怀疑定义一个包含 Cython 类和 sqlalchemy 映射类的包装类的唯一真正答案，并拦截所有属性访问和方法调用以保持它们同步。下面是一个粗略的实现，它应该适用于简单的情况。虽然它几乎没有经过测试，所以几乎可以肯定它有遗漏的错误和极端情况。当心！

def wrapper_class(cls):
    # do this in a function so we can create it generically as needed
    # for each cython class
    class WrapperClass(object):
        def __init__(self,*args,**kwargs):
            # construct the held class using arguments provided
            self._wrapped = cls(*args,**kwargs)

        def __getattribute__(self,name):
            # intercept all requests for attribute access.
            wrapped = object.__getattribute__(self,"_wrapped")
            update_from = wrapped
            update_to = self
            try:
                o = getattr(wrapped,name)
            except AttributeError:
                # if we can't find it look in this class instead.
                # This is reasonable, because there may be methods defined
                # by sqlalchemy for example
                update_from = self
                update_to = wrapped
                o = object.__getattribute__(self,name)
            if callable(o):
                return FunctionWrapper(o,update_from,update_to)
            else:
                return o

        def __setattr__(self,name,value):
            # intercept all attempt to write to attributes
            # and set it in both this class and the wrapped Cython class
            if name!="_wrapped":
                try:
                    setattr(self._wrapped,name,value)
                except AttributeError:
                    pass # ignore errors... maybe bad!
            object.__setattr__(self,name,value)
    return WrapperClass

class FunctionWrapper(object):
    # a problem we have is if we call a member function.
    # It's possible that the member function may change something
    # and thus we need to ensure that everything is updated appropriately
    # afterwards
    def __init__(self,func,update_from,update_to):
        self.__func = func
        self.__update_from = update_from
        self.__update_to = update_to

    def __call__(self,*args,**kwargs):
        ret_val = self.__func(*args,**kwargs)

        # for both Cython classes and sqlalchemy mappings
        # all the relevant attributes exist in the class dictionary
        for k in self.__update_from.__class__.__dict__.iterkeys():
            if not k.startswith('__'): # ignore private stuff
                try:
                    setattr(self.__update_to,k,getattr(self.__update_from,k))
                except AttributeError:
                    # There may be legitmate cases when this fails
                    # (probably relating to sqlalchemy functions?)
                    # in this case, replace raise with pass
                    raise
        return ret_val

要使用它，您可以执行以下操作：

class EquipmentMapped(wrapper_class(Equipment)):
    # you may well have to define __init__ here
    # you'll have to check yourself, and see what sqlalchemy does...
    pass

mapper(EquipmentMapped,equipment)

请记住，这是一个可怕的工作环境，基本上只是在两个地方复制所有数据，然后拼命尝试保持同步。

---

编辑：它的原始版本提供了一种机制来自动执行 OP 尝试过但决定手动执行的查询线（在 Cython 类上定义属性，这只会成功在覆盖 sqlalchemy 的跟踪更改机制）进一步测试证实它不起作用。如果您对不该做什么感到好奇，请查看编辑历史记录！