我创建了一个测试脚本，看起来可以工作。没有实际数据很难判断应该发生什么。我直接使用SA，没有 flask 。而且我刚刚为Parent类创建了一个存根。我使用了新的select()和SQLAlchemy版本1.4/2.0。

import sys
import uuid

from sqlalchemy import (
    create_engine,
    Integer,
    ForeignKey,
    Boolean,
)
from sqlalchemy.schema import (
    Column,
)
from sqlalchemy.sql import select, and_, func, case
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.dialects.postgresql import UUID

Base = declarative_base()

username, password, db = sys.argv[1:4]

engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=True)



class Parent(Base):
    __tablename__ = "parent"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

class ChildVersion(Base):

    __tablename__ = "child"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    deleted = Column(Boolean, default=False, nullable=False)
    parent_id = Column(UUID(as_uuid=True), ForeignKey(Parent.id), nullable=True, index=True)
    subversion = Column(Integer, default=0, nullable=False)
    published = Column(Boolean, default=False, nullable=False)

Base.metadata.create_all(engine)

with Session(engine) as session, session.begin():

    highest_published = select(
        ChildVersion.parent_id.label('parent_id'),
        func.max(ChildVersion.subversion).label('m_version')
    ).select_from(
        ChildVersion
    ).where(
        ChildVersion.published,
        ~ChildVersion.deleted
    ).group_by(
        ChildVersion.parent_id
    ).cte(name='highest_published')

    highest_unpublished = select(
        ChildVersion.parent_id.label('parent_id'),
        func.max(ChildVersion.subversion).label('m_version'),
    ).select_from(
        ChildVersion
    ).where(
        ~ChildVersion.deleted
    ).group_by(
        ChildVersion.parent_id
    ).cte('highest_unpublished')

    max_versions = select(
        case(
            (highest_published.c.parent_id.is_not(None), highest_published.c.parent_id),
            else_=highest_unpublished.c.parent_id).label('parent_id'),
        case(
            (highest_published.c.m_version.is_not(None), highest_published.c.m_version),
            else_=highest_unpublished.c.m_version).label('m_version'),
    ).select_from(
        highest_unpublished
    ).outerjoin(
        highest_published,
        highest_unpublished.c.parent_id==highest_published.c.parent_id,
        # Not in original SQL?
        highest_unpublished.c.m_version==highest_published.c.m_version
    ).cte(name='max_versions')

    versions_with_ids = select(
        ChildVersion.id,
        ChildVersion.parent_id
    ).select_from(
        ChildVersion
    ).join(
        max_versions, and_(
            ChildVersion.parent_id == max_versions.c.parent_id,
            ChildVersion.subversion == max_versions.c.m_version
        )
    ).order_by(ChildVersion.parent_id)

    session.execute(versions_with_ids)

通过SQLAlchemy的echo=True生成SQL。

WITH highest_unpublished AS 
(SELECT child.parent_id AS parent_id, max(child.subversion) AS m_version 
FROM child 
WHERE NOT child.deleted GROUP BY child.parent_id), 
highest_published AS 
(SELECT child.parent_id AS parent_id, max(child.subversion) AS m_version 
FROM child 
WHERE child.published AND NOT child.deleted GROUP BY child.parent_id), 
max_versions AS 
(SELECT CASE WHEN (highest_published.parent_id IS NOT NULL) THEN highest_published.parent_id ELSE highest_unpublished.parent_id END AS parent_id, CASE WHEN (highest_published.m_version IS NOT NULL) THEN highest_published.m_version ELSE highest_unpublished.m_version END AS m_version 
FROM highest_unpublished LEFT OUTER JOIN highest_published ON highest_unpublished.parent_id = highest_published.parent_id)
 SELECT child.id, child.parent_id 
FROM child JOIN max_versions ON child.parent_id = max_versions.parent_id AND child.subversion = max_versions.m_version ORDER BY child.parent_id```

我可以按照错误消息中的建议使用select_from来解决这个问题。

highest_published_version = ChildVersion.query.with_entities(
    ChildVersion.competency_path_id.label('parent_id'),
    sa.func.max(ChildVersion.subversion).label('m_version'),
    ).filter(
        ChildVersion.published, ~ChildVersion.deleted
    ).group_by(ChildVersion.competency_path_id).cte(name='highest_published')

highest_unpublished_version = ChildVersion.query.with_entities(
    ChildVersion.competency_path_id.label('parent_id'),
    sa.func.max(ChildVersion.subversion).label('m_version'), 
    ).filter(
        ~ChildVersion.deleted
    ).group_by(ChildVersion.competency_path_id).cte('highest_unpublished')

versions = db.session.query(highest_published_version, highest_unpublished_version).select_from(highest_unpublished_version).with_entities(
    sa.case(
        (highest_published_version.c.parent_id.is_not(None), highest_published_version.c.parent_id), 
        else_=highest_unpublished_version.c.parent_id).label('parent_id'),
    sa.case(
        (highest_published_version.c.m_version.is_not(None), highest_published_version.c.m_version), 
        else_=highest_unpublished_version.c.m_version).label('m_version'),
).join(highest_published_version, sa.and_(highest_unpublished_version.c.parent_id==highest_published_version.c.parent_id,
            highest_unpublished_version.c.m_version==highest_published_version.c.m_version),
        isouter=True
).cte(name='max_versions')