我正在尝试与一个中间表进行多对多联接。 (首先在Items中选择行,在FK上加入归因,然后在归因中与另一个FK结合以从Items中获得更多信息)模式如下所示:
class Items(Base): __tablename__ = "items" id = Column(Integer, primary_key=True) name = Column(Text) project = Column(Integer) linkid = Column(Integer, ForeignKey("items.linkid")) linked = relationship("Items", foreign_keys="[Items.linkid]") class Attribution(Base): __tablename__ = "attribution" id = Column(Integer, primary_key=True) link_id_d = Column(Integer, ForeignKey('items.linkid')) link_id_m = Column(Integer, ForeignKey('items.linkid'))
查询看起来像这样:
final_items = aliased(Items) proj_1 = session.query(Items)\ .join(Attribution, ( (Items.name=="upper_third") & (Attribution.link_id_m==Items.linkid) ))\ .join(final_items, final_items.linkid==Attribution.link_id_d)\ .all()
这仅提供一行;输出:
upper_third 1
我正在通过sqlite使用的sql,它提供了预期的三行,看起来像这样:
SELECT * FROM items join attribution on items.name = 'upper_third' and items.linkid == attribution.link_id_m join items as tbl1 on tbl1.linkid == attribution.link_id_d
输出:
5|upper_third|1|1|1|2|1|2|lower_first|2|2 5|upper_third|1|1|1|2|1|4|lower_second|2|2 5|upper_third|1|1|1|2|1|6|lower_third|2|2
这两个查询之间的逻辑区别是什么,如何将sql解决方案迁移到sqalchemy?
(用于生成sqlite数据库和测试查询的完整可运行代码在下面]
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, UniqueConstraint, ForeignKey, func, and_
from sqlalchemy.orm import sessionmaker, relationship, aliased, backref
Base = declarative_base()
from sqlalchemy import Column, Integer, String, Float, Boolean, Text, Table
class Items(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True)
name = Column(Text)
project = Column(Integer)
linkid = Column(Integer, ForeignKey("items.linkid"))
linked = relationship("Items", foreign_keys="[Items.linkid]")
attributions_m = relationship("Attribution", foreign_keys="[Attribution.link_id_m]")
attributions_d = relationship("Attribution", foreign_keys="[Attribution.link_id_d]")
class Attribution(Base):
__tablename__ = "attribution"
id = Column(Integer, primary_key=True)
link_id_d = Column(Integer, ForeignKey('items.linkid')) # one of the tests to link to
link_id_m = Column(Integer, ForeignKey('items.linkid')) # one of the tests to link to
import os
if os.path.exists('app.db'):
os.remove('app.db')
engine = create_engine('sqlite+pysqlite:///app.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
u1 = Items(name="upper_first", linkid=1, project=1)
l1 = Items(name="lower_first", linkid=2, project=2)
u2 = Items(name="upper_second", linkid=1, project=1)
l2 = Items(name="lower_second", linkid=2, project=2)
u3 = Items(name="upper_third", linkid=1, project=1)
l3 = Items(name="lower_third", linkid=2, project=2)
session.add(u1)
session.add(l1)
session.add(u2)
session.add(l2)
session.add(u3)
session.add(l3)
session.commit()
a1 = Attribution(link_id_m=u3.linkid, link_id_d=l3.linkid)
session.add(a1)
session.commit()
final_items = aliased(Items)
proj = session.query(Items)\
.join(Attribution, ( (Items.name=="upper_third") & (Attribution.link_id_m==Items.linkid) ))\
.join(final_items, final_items.linkid==Attribution.link_id_d)\
.all()
for l in proj:
print(l.name, l.linkid)
我正在尝试与一个中间表进行多对多联接。 (首先选择“项目”中的行,然后加入FK上的归因,然后与其他FK一起归因以从“项目”中获取更多信息)...
简短的答案:SQLAlchemy正常工作。