如何使用 SQLAlchemy 将一个数据库会话的多对多关系合并到另一个数据库会话？

Question

当我的 python 应用程序启动时，我想从 加载所有数据in.db并将其放入out.db（然后可能在 中进行更改out.db）。我使用session.merge(loaded_object)，但问题是它不保存相关对象。

我的数据是简单的 Person 对象，它们之间具有明显的父子关系（多对多）：

from sqlalchemy import create_engine, Column, String, Integer, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, backref
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Person(Base):
    __tablename__ = "people"
    id = Column(Integer, primary_key=True)
    name = Column(String)

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

    def add_kid(self, kid):
        Edge(kid=kid, parent=self)
        return self

    def get_kids(self):
        return [edge.kid for edge in self.kid_edges]

    def add_parent(self, parent):
        Edge(kid=self, parent=parent)
        return self

    def get_parents(self):
        return [edge.parent for edge in self.parent_edges]

    def __repr__(self):
        return "<Person(id={id}, name={name})>".format(id=str(self.id),
                                                       name=self.name)

class Edge(Base):
    __tablename__ = "edges"
    id = Column(Integer, primary_key=True)
    kid_id = Column(Integer, ForeignKey("people.id"))
    parent_id = Column(Integer, ForeignKey("people.id"))
    kid = relationship("Person", primaryjoin="Edge.kid_id==Person.id",
                       backref=backref("parent_edges", 
                                       collection_class=set))
    parent = relationship("Person", primaryjoin="Edge.parent_id==Person.id",
                          backref=backref("kid_edges",
                                          collection_class=set))

    def __init__(self, kid, parent):
        self.kid = kid
        self.parent = parent

我通过以下方式初始化会话：

db_in_engine = create_engine("sqlite:///in.db", echo=True)
db_in_session_factory = sessionmaker(bind=db_in_engine)
db_in_session = db_in_session_factory()
db_out_engine = create_engine("sqlite:///out.db", echo=True)
db_out_session_factory = sessionmaker(bind=db_out_engine)
db_out_session = db_out_session_factory()
Base.metadata.create_all(db_out_engine)

问题是，当我合并一个人时，孩子们不会合并：

people = db_in_session.query(Person).all()
db_out_session.merge(people[0])
db_out_session.commit() # related Edges, kids and parents of people[0] are not saved

我尝试将cascade="merge" 添加到关系和backrefs 中，但这不起作用。有什么方法可以强制它拯救所有人[0]的孩子/父母和相关边缘？

Answer 1

首先，不要感到难过，因为我必须测试它以了解它为什么不起作用，而且我写了这个东西。

merge() 用例是您从离线缓存或某些本地修改的结构中获取某种应用程序内数据，并将其移动到新会话中的情况。merge() 主要是关于合并更改，因此当它看到没有“更改”的属性时，它假设不需要特殊的工作。因此它会跳过未加载的关系。如果它确实遵循卸载的关系，则合并过程将变得非常缓慢且繁重的操作，因为它会遍历整个关系图并递归地加载所有内容，可能会将数据库的很大一部分加载到内存中以实现高度互连的模式。这里的“从一个数据库复制到另一个数据库”用例是没有预料到的。

如果您只是确保提前加载所有这些边缘，则数据确实会进入，这是一个演示。默认级联也是“保存更新，合并”，因此您不必指定它。

from sqlalchemy import create_engine, Column, String, Integer, ForeignKey
from sqlalchemy.orm import Session, relationship, backref, immediateload
from sqlalchemy.ext.declarative import declarative_base
import os

Base = declarative_base()

class Person(Base):
    __tablename__ = "people"
    id = Column(Integer, primary_key=True)
    name = Column(String)

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


class Edge(Base):
    __tablename__ = "edges"
    id = Column(Integer, primary_key=True)
    kid_id = Column(Integer, ForeignKey("people.id"))
    parent_id = Column(Integer, ForeignKey("people.id"))
    kid = relationship("Person", primaryjoin="Edge.kid_id==Person.id",
                       backref=backref("parent_edges",
                                       collection_class=set))
    parent = relationship("Person", primaryjoin="Edge.parent_id==Person.id",
                          backref=backref("kid_edges",
                                          collection_class=set))

    def __init__(self, kid, parent):
        self.kid = kid
        self.parent = parent

def teardown():
    for path in ("in.db", "out.db"):
        if os.path.exists(path):
            os.remove(path)

def fixture():
    engine = create_engine("sqlite:///in.db", echo=True)
    Base.metadata.create_all(engine)

    s = Session(engine)
    p1, p2, p3, p4, p5 = [Person('p%d' % i) for i in xrange(1, 6)]
    Edge(p1, p2)
    Edge(p1, p3)
    Edge(p4, p3)
    Edge(p5, p2)
    s.add_all([
        p1, p2, p3, p4, p5
    ])
    s.commit()
    return s

def copy(source_session):
    engine = create_engine("sqlite:///out.db", echo=True)
    Base.metadata.create_all(engine)

    s = Session(engine)
    for person in source_session.query(Person).\
            options(immediateload(Person.parent_edges),
                        immediateload(Person.kid_edges)):
        s.merge(person)

    s.commit()

    assert s.query(Person).count() == 5
    assert s.query(Edge).count() == 4

teardown()
source_session = fixture()
copy(source_session)