sqlalchemy join on subquery with new API

Question

I am not able to convert SQL statement into sqlalchemy orm. Issue is I cant join resulting x subquery with FileTag table

I would like to use new API, that is, not using query

SQL statement

    statement = f"""
    SELECT file_tags.id,file_tags.file_id,file_tags.tag,file_tags.category from
    (SELECT  id,file_id, count(DISTINCT tag)
    FROM file_tags
    WHERE {string_condition}
    GROUP BY file_id
    HAVING COUNT(DISTINCT tag) = {num_conditions})  as x
    JOIN file_tags on file_tags.file_id = x.file_id
    """

sqlalchemy ORM (only last part, joining x/file_tag is not working)

    from sqlalchemy import func, subquery, join, alias

    # x is correct and working
    x = (
        select(FileTag)
        .where((FileTag.tag == "eso") & (FileTag.category == "top_category"))
        .group_by(FileTag.file_id)
        .having(func.count(FileTag.tag) == num_conditions)
        .subquery()
    )

    # I am not able to join x subquery. Throws several different errors
    # Here I tried a lot of different variations.
    query = join(x,FileTag.file_id==x.c.file_id)

    session.execute(query)

I think the main issue is that I dont know how to use sqlalchemy, but I tried all possible ways and always getting different errors. Any ideas?

user_2144390 · Answer

Declare x as a .subquery() and then give your join something to join onto:

from sqlalchemy import Column, func, Integer, select, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class FileTag(Base):
    __tablename__ = "file_tags"
    id = Column(Integer, primary_key=True)
    tag = Column(String)
    category = Column(String)
    file_id = Column(String)


num_conditions = 3
x = (
    select(FileTag)
    .where((FileTag.tag == "something") & (FileTag.category == "top_category"))
    .group_by(FileTag.file_id)
    .having(func.count(FileTag.tag) == num_conditions)
    .subquery()
)
query = select(FileTag).join(x, x.c.file_id == FileTag.file_id)
print(query)
"""
SELECT file_tags.id,
    file_tags.tag,
    file_tags.category,
    file_tags.file_id
FROM file_tags
JOIN (
    SELECT file_tags.id AS id,
        file_tags.tag AS tag,
        file_tags.category AS category,
        file_tags.file_id AS file_id
    FROM file_tags
    WHERE file_tags.tag = :tag_1
        AND file_tags.category = :category_1
    GROUP BY file_tags.file_id
    HAVING count(file_tags.tag) = :count_1
    ) AS anon_1
    ON anon_1.file_id = file_tags.file_id
"""

sqlalchemy join on subquery with new API

1 Answers