python - 映射递归关系时出现问题 - SQLAlchemy

Question

我想使用 SQLAlchemy 中的经典映射(数据映射器模式)来获取现有数据模型并将其映射到现有数据库。我已经阅读了与我正在尝试做的事情有关的文档，并且感到迷失，尽管我没有时间按照我想要的方式深入阅读它。我的代码抛出以下错误:

sqlalchemy.exc.NoForeignKeysError: Could not determine join condition between parent/child tables on relationship Person.neighbors - there are no foreign keys linking these tables. Ensure that referencing columns are associated with a ForeignKey or ForeignKeyConstraint, or specify a 'primaryjoin' expression.

我意识到问题在于有两个外键指向同一字段，而 SA 不知道如何选择。我尝试遵循警告消息提供的建议，并根据文档中的示例模拟不同的映射方法，但我尝试的任何方法都不起作用。我正在处理递归关系(即在 ER 图中，这将是与单个实体而不是两个实体相关的多对多关系)。

下面是一些示例代码。人/邻居关系有点奇怪，不是我的实际代码，但它在结构上相似并且代表了我遇到的问题。

from sqlalchemy.orm import mapper, relation
from sqlalchemy import Table, MetaData, Column, Integer, String, ForeignKey
from unittest import TestCase


# Domain Model:
class Person():
    def __init__(self, name, neighbors=[]):
        self.name = name
        self.neighbors = neighbors


class Neighbor():
    def __init__(self, person, distance):
        self.person = person
        self.distance = distance


# Database Schema / ORM objects:
metadata = MetaData()

person = Table(
    'person', metadata,
    Column('name', String(20), primary_key=True))

neighbor = Table(
    'neighbor', metadata,
    Column('name', String(20), ForeignKey(person.c.name), primary_key=True),
    Column('neighbor', String(20), ForeignKey(person.c.name), primary_key=True),
    Column('distance', Integer))


# Data Mappings
def datamap():
    properties = {'neighbors': relation(Neighbor, backref=person)}
    mapper(Person, person, properties=properties)

    properties = {'person': relation(Neighbor, backref=neighbor)}
    mapper(Neighbor, person, properties=properties)


# Unit Test (just to see if we don't throw errors in this case)
class TestMaps(TestCase):
    def test_map(self):
        # Setup
        session = self.get_session()

        # Exercise
        datamap()
        model = self.build_model()
        session.add(model)
        session.flush()

    def build_model(self):
        bill = Person('Bill')
        jack = Person('Jack')
        person.neighbors = [Neighbor(jack, 10)]
        return bill

    def get_session(self):
        from sqlalchemy.orm import sessionmaker
        from sqlalchemy import create_engine
        engine = create_engine('sqlite:///:memory:', echo=False)
        metadata.create_all(engine, tables=[person, neighbor])
        Session = sessionmaker(bind=engine)
        return Session()

Answer 1

您的代码存在太多问题，因此我在这里提供完整的工作副本，并直接在代码中添加注释:

# Domain Model:
class Person(object):
    def __init__(self, name, neighbors=[]):
        self.name = name
        self.neighbors = neighbors


class Neighbor(object):
    def __init__(self, r_neigbor, distance):
        # @NOTE: the first argument is not "person" field, but a real
        # object of the relationship. I gave it name "r_neigbor", but it can
        # be anything, but not a nammed of the mapped attribute ("person")
        self.r_neigbor = r_neigbor
        self.distance = distance


# Database Schema / ORM objects:
metadata = MetaData()

person = Table(
    'person', metadata,
    Column('name', String(20), primary_key=True),
)

neighbor = Table(
    'neighbor', metadata,
    Column('name', String(20), ForeignKey(person.c.name), primary_key=True),
    Column('neighbor', String(20), ForeignKey(person.c.name), primary_key=True),
    Column('distance', Integer),
)


# Data Mappings
def datamap():
    # @NOTE: here it is easier and less confusing to define both relationshis
    # from one side. Again, make sure that the names of the relationships do not
    # override names of existing mapped attributes.
    # @NOTE: `foreign_keys` is the piece which solves the actual error message
    # you get
    properties = {'neighbors':
                  relationship(Neighbor, backref="r_person",
                               foreign_keys=neighbor.c.name,),
                  'neighbors_of':
                  relationship(Neighbor, backref="r_neigbor",
                               foreign_keys=neighbor.c.neighbor,)
                  }
    mapper(Person, person, properties=properties)
    mapper(Neighbor, neighbor)


# Unit Test (just to see if we don't throw errors in this case)
class TestMaps(TestCase):
    def test_map(self):
        # Setup
        session = self.get_session()

        # Exercise
        datamap()
        model = self.build_model()
        session.add(model)
        session.flush()

    def build_model(self):
        bill = Person('Bill')
        jack = Person('Jack')
        # @NOTE: fix: `bill.neighbors` instead of `person.neigbors`
        bill.neighbors = [Neighbor(jack, 10)]
        return bill

    def get_session(self):
        from sqlalchemy.orm import sessionmaker
        from sqlalchemy import create_engine
        engine = create_engine('sqlite:///:memory:', echo=True)
        metadata.create_all(engine, tables=[person, neighbor])
        Session = sessionmaker(bind=engine)
        return Session()