sqlalchemy 使用中间表加入自我（工作 sql，而不是 sqlalchemy）答案

【问题标题】：sqlalchemy join to self with intermediate table (working sql, not sqlalchemy)sqlalchemy 使用中间表加入自我（工作 sql，而不是 sqlalchemy）
【发布时间】：2020-06-21 19:46:23
【问题描述】：

我正在尝试使用中间表执行多对多连接。（首先在 Items 中选择行，在 FK 上加入 Attribution，然后在 Attribution 中加入其他 FK 以从 Items 中获取更多信息）Schema 如下所示：

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解决方案迁移到sqlalchemy？

（生成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)

【问题讨论】：

docs.sqlalchemy.org/en/13/orm/query.htmlall() "将根据主键删除重复条目" PS你研究过吗？在考虑发布之前，请阅读手册和谷歌任何错误消息或您的问题/问题/目标的许多清晰、简洁和精确的措辞，有和没有您的特定字符串/名称和站点：stackoverflow.com 和标签；阅读许多答案。如果您发布问题，请使用一个短语作为标题。请参阅How to Ask 和投票箭头鼠标悬停文本。
@philipxy 问/回答“你研究过吗”这个问题我不认为是一个富有成效的问题。一般来说，通过询问关于 SO 的问题来获取信息对我来说是一种“最后手段”的方法。基本上所有其他方法都更快。所以我在问之前试着用谷歌搜索。但是，基于许多因素，您的搜索词与我的不同，在这种情况下，您可以在文档链接中找到我需要的信息。这就是为什么让第二双眼睛看一眼会有所帮助的原因。如果您发表评论作为答案，我很乐意接受。
你不需要任何帮助来谷歌“sqlalchemy .all()”，这就是我所做的。阅读您正在使用的功能的手册。此外，无论您使用什么，在您的代码问题中输入minimal reproducible example。
基础&结果表内容应该被回显&剪切&粘贴到你的帖子中的原因是1.帖子更有可能包含属于一起的实际输入、代码和输出& 2.帖子自成一体。同样，我建议您从 1 开始显示两个基的 SQL select * & python echo。它显示了查询的实际输入以及初始化代码是正确的 2.它显示了您的 python 读取是正确的。这确定了问题所在。 PS很高兴听到您进行了研究。

标签： python sqlite join sqlalchemy

【解决方案1】：

简答：SQLAlchemy 按预期工作。

关键的罪魁祸首是当您查询映射模型时，您将获得模型的实例作为结果。如果多次返回相同的模型，SA 将确保每个只返回一次，这就是为什么您只返回 1 行而不是预期的 3 行。

您可以通过您构造的 SQL (select * ...) 与 SA 为您创建的 SQL (select items.*；SA 不创建 *，但关键是它仅从items 表。

解决方案是将其他实体也添加到要返回的query(...)：

final_items = aliased(Items, name="FinalItems")
proj = (
    session
    .query(Items, Attribution, final_items)  # IMPORTANT !!!
    .join(Attribution, ( (Items.name=="upper_third") & (Attribution.link_id_m==Items.linkid) ))
    .join(final_items, final_items.linkid==Attribution.link_id_d)
)

运行如下：

for l in proj.all():
    print(l)
    # print(l.Items, l.Attribution, l.FinalItems)  # also can access models using names.

... 将产生tuple(Items, Attribution, Items) 的列表：

(<Items(id=5, linkid=1, name='upper_third', project=1)>, <Attribution(id=1, link_id_d=2, link_id_m=1)>, <Items(id=2, linkid=2, name='lower_first', project=2)>)
(<Items(id=5, linkid=1, name='upper_third', project=1)>, <Attribution(id=1, link_id_d=2, link_id_m=1)>, <Items(id=4, linkid=2, name='lower_second', project=2)>)
(<Items(id=5, linkid=1, name='upper_third', project=1)>, <Attribution(id=1, link_id_d=2, link_id_m=1)>, <Items(id=6, linkid=2, name='lower_third', project=2)>)

如果您真的想要返回示例中的列，您可以执行查询的statement。下面的代码

for row in session.execute(proj.statement):
    print(row)

将返回：

(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)

【讨论】：