麻烦映射递归关系 - SQLAlchemy
Trouble Mapping Recursive Relationship - SQLAlchemy
我想使用 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 图中，这将是与单个实体相关的多对多关系，而不是两个）。
下面是一些示例代码。 Person/Neighbor 关系有点奇怪，不是我的实际代码，但它在结构上相似，代表了我遇到的问题。
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()
您的代码存在太多问题，我在这里直接在代码中提供完整的工作副本和注释：
# 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()
这个答案只是配置映射的另一种方式，如对已接受答案的评论中所问：
# Domain Model:
class Person(object):
def __init__(self, name):
self.name = name
# 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': relationship(
Person, backref="neighbor_of", secondary=neighbor,
primaryjoin=person.c.name == neighbor.c.name,
secondaryjoin=person.c.name == neighbor.c.neighbor,
),
}
mapper(Person, person, properties=properties)
class TestMaps(TestCase):
def test_map(self):
# Setup
session = self.get_session()
# Exercise
datamap()
model = self.build_model()
session.add(model)
session.flush()
# Anna moved in
model.neighbors.append(Person('Anna'))
session.commit()
def build_model(self):
bill = Person('Bill')
jack = Person('Jack')
bill.neighbors.append(jack)
assert bill in jack.neighbor_of
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()
我想使用 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 图中，这将是与单个实体相关的多对多关系，而不是两个）。
下面是一些示例代码。 Person/Neighbor 关系有点奇怪，不是我的实际代码，但它在结构上相似，代表了我遇到的问题。
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()
您的代码存在太多问题，我在这里直接在代码中提供完整的工作副本和注释：
# 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()
这个答案只是配置映射的另一种方式，如对已接受答案的评论中所问：
# Domain Model:
class Person(object):
def __init__(self, name):
self.name = name
# 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': relationship(
Person, backref="neighbor_of", secondary=neighbor,
primaryjoin=person.c.name == neighbor.c.name,
secondaryjoin=person.c.name == neighbor.c.neighbor,
),
}
mapper(Person, person, properties=properties)
class TestMaps(TestCase):
def test_map(self):
# Setup
session = self.get_session()
# Exercise
datamap()
model = self.build_model()
session.add(model)
session.flush()
# Anna moved in
model.neighbors.append(Person('Anna'))
session.commit()
def build_model(self):
bill = Person('Bill')
jack = Person('Jack')
bill.neighbors.append(jack)
assert bill in jack.neighbor_of
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()