SQLAlchemy：从另一个查询的结果中查找表中的所有匹配行

Posted 2023-03-16

【中文标题】SQLAlchemy：从另一个查询的结果中查找表中的所有匹配行

【英文标题】：SQLAlchemy: Finding all matching rows from a table with results from another query

【发布时间】：2022-01-16 22:06:06

【问题描述】：

我有四张桌子：

页面 wikiPageID 网址 收藏 collectionID wikiPageID PageReference（连接表） wikiPageID 参考ID 参考 参考ID

对于与page 表中的行对应的任何给定 URL，我想返回分配给可能属于同一集合的任何同级页面的所有引用。

我觉得可能有更有效的方法来做到这一点。到目前为止，我已将其分解为一组查询，并且只完成了一半。

    # Get all ReferenceData from Reference table for ID
    wikiPageID = ""
    url = request.json['url']

    # Get WikiPageID from Page table where Page.url = request.json['url']
    for item in db.session.query(PageModel).filter(PageModel.url == url).values('wikiPageID'):
        wikiPageID = item[0]
        #print(wikiPageID)

    for item in 
    db.session.query(CollectionModel).filter(CollectionModel.wikiPageID == wikiPageID).values('collectionID'):
        collectionID = item[0]
        #print(collectionID)

    # Get all WikiPageID's from Collection table where WikiPageID EXISTS
    result1 = db.session.query(CollectionModel).filter(CollectionModel.collectionID == collectionID).values('wikiPageID')

    ## THIS IS WHERE IT STARTS GOING PEAR SHAPED
    # Get all ReferenceID from PageReference where WikiPageID EXISTS
    result = db.session.query(PageReferenceModel).filter(PageReferenceModel.wikiPageID.in_(result1) )

    schema = CollectionSchema()
    return schema.dump(result, many=True), 200

在我的最后一个查询中，我试图嵌套之前的查询结果。当我试图通过从先前结果​​传递的 ID 过滤结果时。我觉得我做错了，不知道从哪里开始。

SQLAlchemy 模型

class CollectionModel(db.Model):
    __tablename__ = "Collection"

    collectionID = db.Column(db.Integer, primary_key=True)
    wikiPageID = db.Column(db.Integer, db.ForeignKey('Page.wikiPageID'), nullable=False)
    Page = db.relationship("PageModel")

class PageModel(db.Model):
    __tablename__ = "Page"

    wikiPageID = db.Column(db.Integer, primary_key=True)
    langCode = db.Column(db.String, nullable=True)
    title = db.Column(db.String, nullable=True)
    url = db.Column(db.String, nullable=True)
    summary = db.Column(db.String, nullable=True)
    lastUpdated = db.Column(db.String, nullable=True)

class PageReferenceModel(db.Model):
    __tablename__ = "PageReference"

    pageReferenceID = db.Column(db.Integer, primary_key=True)
    wikiPageID = db.Column(db.Integer, db.ForeignKey('Page.wikiPageID'), nullable=False)
    referenceID = db.Column(db.Integer, db.ForeignKey('Reference.referenceID'), nullable=False)
    Page = db.relationship("PageModel")
    Reference = db.relationship("ReferenceModel")


class ReferenceModel(db.Model):
    __tablename__ = "Reference"

    referenceID = db.Column(db.Integer, primary_key=True)
    langCode = db.Column(db.String, nullable=False)
    isGov = db.Column(db.String, nullable=False)
    type = db.Column(db.String, nullable=True)
    title = db.Column(db.String, nullable=True)
    domain = db.Column(db.String, nullable=True)
    suffix = db.Column(db.String, nullable=True)
    description = db.Column(db.String, nullable=True)
    publishDate = db.Column(db.String, nullable=True)
    url = db.Column(db.String, nullable=True)
    youtubeId = db.Column(db.String, nullable=True)
    langCheck = db.Column(db.Boolean, nullable=True)

【参考方案1】：

已修复：

我最终对先前查询的结果执行了一系列 if 语句。仍然很可能效率低下。但它有效。

收集 GET

    # Get all ReferenceData from Reference table for ID
    wikiPageID = ""
    url = request.json['url']


   # Get all ReferenceData from Reference table for ID
    wikiPageID = ""
    url = request.json['url']

    # Get WikiPageID from Page table where Page.url = request.json['url']
    for item in db.session.query(PageModel).filter(PageModel.url == url).values(PageModel.wikiPageID):
        wikiPageID = item[0]
        #print(wikiPageID)

    for item in db.session.query(CollectionModel).filter(CollectionModel.wikiPageID == wikiPageID).values(CollectionModel.collectionID):
        collectionID = item[0]
        #print(collectionID)

    # Get all WikiPageID's from Collection table where WikiPageID EXISTS
    wikiPageIDResult = db.session.query(CollectionModel).filter(CollectionModel.collectionID == collectionID).values(CollectionModel.wikiPageID)

    wikiPageArray = []
    for item in wikiPageIDResult:
        wikiPageArray.append(item[0])
        print("Wiki Page Array: ", wikiPageArray)

    #Get all ReferenceID from PageReference where WikiPageID EXISTS
    wikiPageReferenceResults = db.session.query(PageReferenceModel).filter(PageReferenceModel.wikiPageID.in_(wikiPageArray)).values(PageReferenceModel.referenceID)

    wikiPageReferenceArray = []
    for item in wikiPageReferenceResults:
        wikiPageReferenceArray.append(item[0])
        print("Wiki Page Reference Array: ", wikiPageReferenceArray)

    referenceResult = db.session.query(ReferenceModel).filter(ReferenceModel.referenceID.in_(wikiPageReferenceArray))

    schema = ReferenceSchema()
    return schema.dump(referenceResult, many=True), 200

【参考方案2】：

你应该使用join：

    result = db.session.query(PageReferenceModel)\
             .join(CollectionModel, CollectionModel.wikiPageId == PageReferenceModel.wikiPageId)\
             .join(PageModel, PageModel.wikiPageId == PageReferenceModel.wikiPageId)\
             .filter(PageModel.url == url)

    schema = CollectionSchema()
    return schema.dump(result, many=True), 200

【讨论】：

整洁多了。但是我收到以下错误：sqlalchemy.exc.InvalidRequestError: Don't know how to join to <class '__main__.CollectionModel'>. Please use the .select_from() method to establish an explicit left side, as well as providing an explcit ON clause if not present already to help resolve the ambiguity.

如果你在这里展示你的sqlalchemy模型定义会更容易

我已经添加了这些。

我修正了连接定义中的错字，请立即尝试@TheMightyLlama

这行得通，虽然它没有让我达到我的目标。它返回[ "wikiPageID": 1 ]，我正在寻找的是集合中存在的所有引用，其中包含一个带有url x [ "publishDate": "15/09/1980", "langCode": "EN", "suffix": ".gov.uk", "youtubeId": "", "type": "article", "description": "ssdfsdfsdf", "url": "asdfsdfsd.com", "langCheck": false, "isGov": "1", "domain": "government", "title": "Article summary" ]

的页面