python - 在 sqlalchemy 映射集合中使用值元组

Question

在多对多关系中，我在关联表上有一些额外的数据来描述关系(一个数量和一个 bool 值)。我想使用映射集合来避免直接使用关联对象，但我无法弄清楚如何将元组用于映射中的值。据我所知，Attribute as dict of lists using middleman table with SQLAlchemy相似，但倒退。

为了说明这一点，我想做这样的事情:

>>> collection.items[item] = (3, True)
>>> collection.items[item] = (1, False)
>>> colletion.items
{"item name": (3, True), "item name": (1, False)}

这...有效...但最终 SQLAlchemy 会尝试将元组放入数据库(稍后我会尝试重新创建)。

我也曾尝试在键(相关对象和其他列之一)中使用元组，但它看起来很糟糕，并且它不起作用:

>>> collection.items[item, True]  = 3
>>> collection.items[item, False] = 1
>>> collection.items
{(<item>, True): 3, (<item>, False): 1}

我可以毫无问题地将项目名称和一个值放入映射集合中:我有这种关系的另一种(结构相同的)形式，我通过建立两个关系(和关联代理)来解决它根据 bool 值划分它们之间的关联表，并且它们的创建者函数正确设置 bool 值而没有任何进一步的干扰。不幸的是，在那种情况下， bool 值指定了一个较小的语义差异(应用程序代码需要将项目视为一个组)，而在当前问题中，这是一个不小的外观差异(应用程序代码不应将项目视为组，但值确实会改变项目的显示方式，因此是必需的)。

Answer 1

链接的答案包含所有组件。 attribute_mapped_collection 和 association_proxy 可以一起做很多事情。首先是 string->tuple(int, boolean) 的字典(为 m2m 更新):

from sqlalchemy import Integer, Boolean, String, Column, create_engine, \
    ForeignKey
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm.collections import attribute_mapped_collection

Base = declarative_base()

class SomeClass(Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    tuple_elements = relationship(
                "TupleAssociation", 
                collection_class=attribute_mapped_collection("name"),
                cascade="all, delete-orphan"
            )
    items = association_proxy("tuple_elements", "as_tuple")

class TupleAssociation(Base):
    __tablename__ = 'tuple_association'
    parent_id = Column(Integer, ForeignKey('sometable.id'), primary_key=True)
    tuple_id = Column(Integer, ForeignKey("tuple_data.id"), primary_key=True)
    name = Column(String)

    tuple_element = relationship("TupleElement")

    def __init__(self, key, tup):
        self.name = key
        self.tuple_element = TupleElement(tup)

    @property
    def as_tuple(self):
        return self.tuple_element.as_tuple

class TupleElement(Base):
    __tablename__ = 'tuple_data'

    id = Column(Integer, primary_key=True)
    col1 = Column(Integer)
    col2 = Column(Boolean)

    def __init__(self, tup):
        self.col1, self.col2 = tup

    @property
    def as_tuple(self):
        return self.col1, self.col2


e = create_engine('sqlite://')
Base.metadata.create_all(e)
s = Session(e)

collection = SomeClass()
collection.items["item name 1"] = (3, True)
collection.items["item name 2"] = (1, False)
print collection.items

s.add(collection)
s.commit()

collection = s.query(SomeClass).first()
print collection.items

这是关联上的元组和端点上的名称的另一种方式:

from sqlalchemy import Integer, Boolean, String, Column, create_engine, \
    ForeignKey
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm.collections import attribute_mapped_collection

Base = declarative_base()

class SomeClass(Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    tuple_elements = relationship(
                "TupleAssociation", 
                collection_class=attribute_mapped_collection("name"),
                cascade="all, delete-orphan"
            )
    items = association_proxy("tuple_elements", "as_tuple")

class TupleAssociation(Base):
    __tablename__ = 'tuple_association'
    parent_id = Column(Integer, ForeignKey('sometable.id'), primary_key=True)
    name_id = Column(Integer, ForeignKey("name_data.id"), primary_key=True)

    col1 = Column(Integer)
    col2 = Column(Boolean)

    name_element = relationship("NameElement")

    def __init__(self, key, tup):
        self.name_element = NameElement(name=key)
        self.col1, self.col2 = tup

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

    @property
    def as_tuple(self):
        return self.col1, self.col2

class NameElement(Base):
    __tablename__ = 'name_data'

    id = Column(Integer, primary_key=True)
    name = Column(String)


e = create_engine('sqlite://', echo=True)
Base.metadata.create_all(e)
s = Session(e)

collection = SomeClass()
collection.items["item name 1"] = (3, True)
collection.items["item name 2"] = (1, False)
print collection.items

s.add(collection)
s.commit()

collection = s.query(SomeClass).first()
print collection.items

这可能就是您所需要的。如果您使用的是支持 SQL 元组的 Postgresql，您可以使用混合体加上 tuple_() 在上面添加更多内容，以便 as_tuple 可以在 SQL 级别使用也(下面也使用一对多而不是关联对象来举例):

from sqlalchemy import Integer, Boolean, String, Column, create_engine, \
    ForeignKey
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.ext import hybrid
from sqlalchemy.sql import tuple_

Base = declarative_base()

class SomeClass(Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    tuple_elements = relationship(
                "TupleElement", 
                collection_class=attribute_mapped_collection("name"),
                cascade="all, delete-orphan"
            )
    items = association_proxy("tuple_elements", "as_tuple")

class TupleElement(Base):
    __tablename__ = 'tuple_data'

    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('sometable.id'), nullable=False)
    name = Column(String)
    col1 = Column(Integer)
    col2 = Column(Boolean)

    def __init__(self, key, tup):
        self.name = key
        self.col1, self.col2 = tup

    @hybrid.hybrid_property
    def as_tuple(self):
        return self.col1, self.col2

    @as_tuple.expression
    def as_tuple(self):
        return tuple_(self.col1, self.col2)

e = create_engine('postgresql://scott:tiger@localhost/test', echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)
s = Session(e)

collection = SomeClass()
collection.items["item name 1"] = (3, True)
collection.items["item name 2"] = (1, False)
print collection.items

s.add(collection)
s.commit()

q = s.query(SomeClass).join(SomeClass.tuple_elements)
assert q.filter(TupleElement.as_tuple == (3, True)).first() is collection
assert q.filter(TupleElement.as_tuple == (5, False)).first() is None
print s.query(TupleElement.as_tuple).all()