在SQLAlchemy映射集合中使用值元组

Question

在多对多关系中，我在关联表上有一些额外的数据，用来描述这种关系（比如数量和一个布尔值）。我想使用映射集合，这样就不需要直接处理关联对象了，但我不知道怎么用元组来表示映射中的值。根据我所了解的，使用中间表的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}

我可以在映射集合中放入项目名称和一个值，这没有问题：我还有另一种（结构上相同的）关系，通过创建两个关系（和关联代理）来解决这个问题，它们根据布尔值将关联表分开，而它们的创建函数会正确设置布尔值，没有进一步的干扰。不幸的是，在那个情况下，布尔值指定了一个小的语义差异（应用代码需要将这些项目视为一组），而在当前的问题中，这是一种不小的外观差异（应用代码不应该将这些项目视为组，但这个值确实会影响项目的显示，因此是必要的）。

Answer 1

链接中的回答包含了所有的组成部分。attribute_mapped_collection和association_proxy可以一起做很多事情。首先，这里有一个字符串到元组(int, boolean)的字典（针对多对多关系进行了更新）：

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

这可能就是你所需要的全部。如果你使用的是Postgresql，它支持SQL元组，你可以在上面的基础上添加更多内容，使用混合类型加上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()