Làm thế nào để tích hợp SQLAlchemy và một Numpy.ndarray lớp con trơn tru và theo một cách nhiệt tình?

Question

Tôi muốn lưu trữ các mảng numpy với các chú thích (như name) thông qua SQLAlchemy trong một cơ sở dữ liệu quan hệ. Để làm như vậy,

• tôi tách mảng NumPy từ dữ liệu của nó thông qua một đối tượng dữ liệu transfere (DTONumpy như một phần của MyNumpy).
• đối tượng khó hiểu được thu thập với Container.

Điều gì sẽ là một cách tốt đẹp và pythonic để sửa đổi Container (từ ví dụ dưới đây) theo một cách mà nó cung cấp như một danh sách trực tiếp MyNumpy -objects thay vì DTONumpy được cung cấp bởi SQLAlchemy?

Dưới đây là một minh họa về các vấn đề:

import numpy as np 
import zlib 

import sqlalchemy as sa 
from sqlalchemy.orm import relationship, scoped_session, sessionmaker 
from sqlalchemy.ext.declarative import declarative_base 
from sqlalchemy.types import TypeDecorator, CHAR 

DBSession = scoped_session(sessionmaker()) 
Base = declarative_base() 

#### New SQLAlchemy-Type ##################### 
class NumpyType (sa.types.TypeDecorator): 
    impl = sa.types.LargeBinary 

    def process_bind_param(self, value, dialect): 
    return zlib.compress(value.dumps(), 9) 

    def process_result_value(self, value, dialect): 
    return np.loads(zlib.decompress(value)) 
############################################## 


class DTONumpy(Base): 
    __tablename__ = 'dtos_numpy' 
    id = sa.Column(sa.Integer, primary_key=True) 
    amount = sa.Column('amount', NumpyType) 
    name = sa.Column('name', sa.String, default='') 
    container_id = sa.Column(sa.ForeignKey('containers.id')) 

    container_object = relationship(
     "Container", 
     uselist=False, 
     backref='dto_numpy_objects' 
    ) 

    def __init__(self, amount, name=None): 
    self.amount = np.array(amount) 
    self.name = name 


class Container(Base): 
    __tablename__ = 'containers' 
    id = sa.Column(sa.Integer, primary_key=True) 
    name = sa.Column(sa.String, unique=True) 

    # HERE: how to access DTONumpy BUT as MyNumpy-Objects in a way that MyNumpy 
    # is smoothly integrated into SQLAlchemy 


class MyNumpy(np.ndarray): 
    _DTO = DTONumpy 
    def __new__(cls, amount, name=''): 
    dto = cls._DTO(amount=amount, name=name) 
    return cls.newByDTO(dto) 

    @classmethod 
    def newByDTO(cls, dto): 
    obj = np.array(dto.amount).view(cls) 
    obj.setflags(write=False) # Immutable 
    obj._dto = dto 
    return obj 

    @property 
    def name(self): 
    return self._dto.name 


if __name__ == '__main__': 
    engine = sa.create_engine('sqlite:///:memory:', echo=True) 
    DBSession.configure(bind=engine) 
    Base.metadata.create_all(engine) 
    session = DBSession() 

    mn1 = MyNumpy ([1,2,3], "good data") 
    mn2 = MyNumpy ([2,3,4], "bad data") 

    # Save MyNumpy-Objects 
    c1 = Container() 
    c1.name = "Test-Container" 
    c1.dto_numpy_objects += [mn1._dto, mn2._dto] # not a good ui 
    session.add(c1) 
    session.commit() 

    # Load MyNumpy-Objects 
    c2 = session.query(Container).filter_by(name="Test-Container").first() 
    # Ugly UI: 
    mn3 = MyNumpy.newByDTO(c2.dto_numpy_objects[0]) 
    mn4 = MyNumpy.newByDTO(c2.dto_numpy_objects[1]) 
    name3 = mn3._dto.name 
    name4 = mn4._dto.name 

Container bây giờ sẽ cung cấp một danh sách các MyNumpy -objects và MyNumpy -object một tham chiếu đến Container -object theo (danh sách và tài liệu tham khảo sẽ phải lấy SQLAlchemy-mapping vào tài khoản):

type (c2.my_numpy_objects[0]) == MyNumpy 
>>> True 
c2.my_numpy_objects.append(MyNumpy ([7,2,5,6], "new data") 
print c2.dto_numpy_objects[-1].name 
>>> "new data" 

Answer 1

Sử dụng ListView -answer từ that câu hỏi, tôi đã đưa ra fo giải pháp llowing:

Thứ nhất, sửa đổi Container bằng cách thêm một ListView -property trên đầu trang của SQLAlchemy-tài sản dto_numpy_objects:

def __init__(self, name): 
    self.name = name 
    """ 
    At this point, the following code doesn't work: 
    --------------------- 
    self.my_numpies = ListView(
     self.dto_numpy_objects, # see `DTO_Numpy.container_object` 
     MyNumpy.newByDTO, 
     MyNumpy.getDTO) 
    --------------------- 
    SQLAlchemy seems to change the `dto_numypy_object`-object after the 
    init-call. Thus, `my_numpies._data` doesn't reference `dto_numpy_objects` 
    anymore. One solution is to implement a property that initalizes `ListView` 
    on first access. See below, property `Container.my_numpies`. 
    """ 

    @property 
    def my_numpies(self): 
    if not hasattr(self, '_my_numpies'): 
     # The following part can not be exe 
     self._my_numpies = ListView(
      self.dto_numpy_objects, # see `DTO_Numpy.container_object` 
      MyNumpy.newByDTO, 
      MyNumpy.getDTO) 

    return self._my_numpies 

Thứ hai, thêm phương pháp getDTO mà có thể được sử dụng như new2raw-chuyển đổiMyNumpy:

def getDTO(self): 
    return self._dto 

Để sử dụng backrefcontainer_object cũng từ MyNumpy thực hiện nó như là một wrapper bằng cách thêm các phương pháp sau đây:

def __getattr__(self, attr): 
    return getattr(self._dto, attr) 

Tất cả cùng nhau, mã trông như thế này:

import numpy as np 
import zlib 

import sqlalchemy as sa 
from sqlalchemy.orm import relationship, scoped_session, sessionmaker 
from sqlalchemy.ext.declarative import declarative_base 
from sqlalchemy.types import TypeDecorator, CHAR 

DBSession = scoped_session(sessionmaker()) 
Base = declarative_base() 


class ListView(list): 
    def __init__(self, raw_list, raw2new, new2raw): 
    self._data = raw_list 
    self.converters = {'raw2new': raw2new, 
     'new2raw': new2raw} 

    def __repr__(self): 
    repr_list = [self.converters['raw2new'](item) for item in self._data] 
    repr_str = "[" 
    for element in repr_list: 
     repr_str += element.__repr__() + ",\n " 
    repr_str = repr_str[:-3] + "]" 
    return repr_str 

    def append(self, item): 
    self._data.append(self.converters['new2raw'](item)) 

    def pop(self, index): 
    self._data.pop(index) 

    def __getitem__(self, index): 
    return self.converters['raw2new'](self._data[index]) 

    def __setitem__(self, key, value): 
    self._data.__setitem__(key, self.converters['new2raw'](value)) 

    def __delitem__(self, key): 
    return self._data.__delitem__(key) 

    def __getslice__(self, i, j): 
    return ListView(self._data.__getslice__(i,j), **self.converters) 

    def __contains__(self, item): 
    return self._data.__contains__(self.converters['new2raw'](item)) 

    def __add__(self, other_list_view): 
    assert self.converters == other_list_view.converters 
    return ListView(
     self._data + other_list_view._data, 
     **self.converters) 

    def __len__(self): 
    return len(self._data) 

    def __iter__(self): 
    return iter([self.converters['raw2new'](item) for item in self._data]) 

    def __eq__(self, other): 
    return self._data == other._data 


#### New SQLAlchemy-Type ##################### 
class NumpyType (sa.types.TypeDecorator): 
    impl = sa.types.LargeBinary 

    def process_bind_param(self, value, dialect): 
    return zlib.compress(value.dumps(), 9) 

    def process_result_value(self, value, dialect): 
    return np.loads(zlib.decompress(value)) 
############################################## 


class DTONumpy(Base): 
    __tablename__ = 'dtos_numpy' 
    id = sa.Column(sa.Integer, primary_key=True) 
    amount = sa.Column('amount', NumpyType) 
    name = sa.Column('name', sa.String, default='') 
    container_id = sa.Column(sa.ForeignKey('containers.id')) 

    container_object = relationship(
     "Container", 
     uselist=False, 
     backref='dto_numpy_objects' 
    ) 

    def __init__(self, amount, name=None): 
    self.amount = np.array(amount) 
    self.name = name 

    def reprInitParams(self): 
    return "(%r, %r)" %(self.amount, self.name) 

    def __repr__(self): 
    return "%s%s" %(
     self.__class__.__name__, 
     self.reprInitParams()) 


class Container(Base): 
    __tablename__ = 'containers' 
    id = sa.Column(sa.Integer, primary_key=True) 
    name = sa.Column(sa.String, unique=True) 

    def __init__(self, name): 
    self.name = name 
    super(Container, self).__init__() 

    @property 
    def my_numpies(self): 
    if not hasattr(self, '_my_numpies'): 
     # The following part can not be exe 
     self._my_numpies = ListView(
      self.dto_numpy_objects, # see `DTO_Numpy.container_object` 
      MyNumpy.newByDTO, 
      MyNumpy.getDTO) 

    return self._my_numpies 


class MyNumpy(np.ndarray): 
    _DTO = DTONumpy 
    def __new__(cls, amount, name=''): 
    dto = cls._DTO(amount=amount, name=name) 
    return cls.newByDTO(dto) 

    @classmethod 
    def newByDTO(cls, dto): 
    obj = np.array(dto.amount).view(cls) 
    obj.setflags(write=False) # Immutable 
    obj._dto = dto 
    return obj 

    @property 
    def name(self): 
    return self._dto.name 

    def getDTO(self): 
    return self._dto 

    def __getattr__(self, attr): 
    return getattr(self._dto, attr) 

    def __repr__(self): 
    return "%s%s" %(
     self.__class__.__name__, 
     self._dto.reprInitParams()) 


if __name__ == '__main__': 
    engine = sa.create_engine('sqlite:///:memory:', echo=True) 
    DBSession.configure(bind=engine) 
    Base.metadata.create_all(engine) 
    session = DBSession() 

    mn1 = MyNumpy ([1,2,3], "good data") 
    mn2 = MyNumpy ([2,3,4], "bad data") 

    # Save MyNumpy-Objects 
    c1 = Container("Test-Container") 
    c1.my_numpies.append(mn1) 
    c1.my_numpies.append(mn2) 
    session.add(c1) 
    session.commit() 

    # Load MyNumpy-Objects 
    c2 = session.query(Container).filter_by(name="Test-Container").first() 
    mn3 = c1.my_numpies[0] 
    mn4 = c1.my_numpies[1] 

Đối với đại diện tốt hơn tôi đã thêm

• DTONumpy.reprInitParams
• DTONumpy.__repr__
• MyNumpy.__repr__

Một điều mà vẫn không hoạt động:

c1.my_numpies += [mn1, mn2.dto]