如何使用关联对象和辅助连接构建反向连接

Question

我需要一些模型，例如:

• 作品-例如文学作品。
• 工作者-例如作曲家，翻译或类似的人对工作有贡献。

因此，需要一个'type'字段来通过分工来区分工人。作为SQLAlchemy的文档，这种情况可以受益于关联对象，如下所示:

class Work(base):
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    description = Column(Text)
class Worker(base):
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    description = Column(Text)
class Assignment(base):
    work_id = Column(Integer, Foreignkey('work.id'), primary_key=True)
    worker_id = Column(Integer, Foreignkey('worker.id'), primary_key=True)
    type = Column(SmallInteger, nullable=True)

然而，如何利用backref和的可选连接条件来立即建立关系，实现每个Work对象都可以通过不同的属性来检索和修改相应的Worker (s)。例如:

work = session.query(Work).get(1)
work.name
>>> 'A Dream of The Red Mansions'
work.composers
>>> [<Worker('Xueqin Cao')>]
work.translators
>>> [<Worker('Xianyi Yang')>, <Worker('Naidie Dai')>]

亦然:

worker = session.query(Worker).get(1)
worker.name
>>> 'Xueqin Cao'
worker.composed
>>> [<Work('A Dream of The Red Mansions')>]
worker.translated
>>> []

直接添加secondaryjoin而不指定secondary似乎是不可行的，此外，SQLAlchemy的文档指出:

当使用关联对象模式时，建议不要将关联映射表用作其他地方关系()的次要参数，除非该关系()包含选项viewonly=True。否则，如果在相关属性和相关对象上检测到类似的状态，SQLAlchemy可能会尝试在同一表上发出冗余的INSERT和DELETE语句。

那么，有没有一种方法可以既优雅又容易地建立这些关系呢?

Answer 1

一般有三种方法。

一个是，做一个"香草"设置，在没有区分"类型"的情况下设置"工作"/"工人"-然后，使用relationship()为"作曲家"，"组成"，"翻译器"，"翻译"通过使用"次要"到Assignment.__table__以及自定义连接条件，以及viewonly=True。所以你只能通过香草属性来写。这里的缺点是"普通"集合和"特定"集合之间没有立即同步。

另一个是，与"香草"设置相同，但只是使用普通的Python描述符在内存中给出"composer"，"compose"，"translator"，"translned"视图，即[obj.worker for obj in self.workers if obj.type == 'composer']。这是最简单的方法。无论您在"香草"集合中放入什么，都会在"过滤"集合中显示出来，SQL很简单，并且有更少的SELECT语句(每个Worker/Work一个而不是每个Worker/Work N个)。

最后，最接近你所要求的方法，使用主连接和反向引用，但注意与关联对象，反向引用是在工作/分配和分配/工作之间，而不是直接在工作/工作之间。这种方法最终可能会使用更多的SQL来获得结果，但它是最完整的，并且还具有自动编写"类型"的漂亮特性。我们也使用了"单向回退"，因为赋值并没有一个简单的向外关联的方法(有办法做到这一点，但它会很乏味)。使用Python函数自动创建关系可以减少样板文件，注意这里我使用的是字符串"type"，如果您向系统添加更多参数，它可以是整数:

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy
Base = declarative_base()
def _work_assignment(name):
    assign_ = relationship("Assignment",
                    primaryjoin="and_(Assignment.work_id==Work.id, "
                                    "Assignment.type=='%s')" % name,
                    back_populates="work", cascade="all, delete-orphan")
    assoc = association_proxy("%s_assign" % name, "worker",
                    creator=lambda worker: Assignment(worker=worker, type=name))
    return assoc, assign_
def _worker_assignment(name):
    assign_ = relationship("Assignment",
                    primaryjoin="and_(Assignment.worker_id==Worker.id, "
                                    "Assignment.type=='%s')" % name,
                    back_populates="worker", cascade="all, delete-orphan")
    assoc = association_proxy("%s_assign" % name, "work",
                    creator=lambda work: Assignment(work=work, type=name))
    return assoc, assign_
class Work(Base):
    __tablename__ = 'work'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    description = Column(Text)
    composers, composer_assign = _work_assignment("composer")
    translators, translator_assign = _work_assignment("translator")
class Worker(Base):
    __tablename__ = 'worker'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    description = Column(Text)
    composed, composer_assign = _worker_assignment("composer")
    translated, translator_assign = _worker_assignment("translator")
class Assignment(Base):
    __tablename__ = 'assignment'
    work_id = Column(Integer, ForeignKey('work.id'), primary_key=True)
    worker_id = Column(Integer, ForeignKey('worker.id'), primary_key=True)
    type = Column(String, nullable=False)
    worker = relationship("Worker")
    work = relationship("Work")
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

session = Session(e)
ww1, ww2, ww3 = Worker(name='Xueqin Cao'), Worker(name='Xianyi Yang'), Worker(name='Naidie Dai')
w1 = Work(name='A Dream of The Red Mansions')
w1.composers.append(ww1)
w1.translators.extend([ww2, ww3])
session.add(w1)
session.commit()
work = session.query(Work).get(1)
assert work.name == 'A Dream of The Red Mansions'
assert work.composers == [ww1]
assert work.translators == [ww2, ww3]
worker = session.query(Worker).get(ww1.id)
assert worker.name == 'Xueqin Cao'
assert worker.composed == [work]
assert worker.translated == []
worker.composed[:] = []
# either do this...
session.expire(work, ['composer_assign'])
# or this....basically need composer_assign to reload
# session.commit()
assert work.composers == []