Python Sqlalchemy Postgres 如何存储子查询结果避免重复执行

核心方法：使用CTE（公用表达式）复用子查询 #

在SQLAlchemy中可以通过.cte()方法将重复使用的子查询定义为公用表达式，PostgreSQL在执行时只会对CTE计算一次结果，后续所有引用都会复用该结果，完美解决子查询重复执行问题。

修改后代码示例 #

仅需要在原有代码基础上做少量调整：

from sqlalchemy.sql.schema import ForeignKey
from sqlalchemy import Column, Integer, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import select, union

Base = declarative_base()
class Table1(Base):
    __tablename__ = 'table1'
    id = Column(Integer, primary_key=True)
    uuid = Column(Text, unique=True, nullable=False)
    
class Table2(Base):
    __tablename__ = 'table2'
    id = Column(Integer, primary_key=True)
    uuid = Column(Text, unique=True, nullable=False)
    
class Table3(Base):
    __tablename__ = 'table3'
    id = Column(Integer, primary_key=True)
    uuid = Column(Text, unique=True, nullable=False)
    
class Table4(Base):
    __tablename__ = 'table4'
    id = Column(Integer, primary_key=True)
    type = Column(Text, nullable=False)

class Table5(Base):
    __tablename__ = 'table5'
    id = Column(Integer, primary_key=True)
    res_id = Column(Integer, ForeignKey('table4.id'), nullable=False)
    value = Column(Text, nullable=False)

class Table1Map(Base):
    __tablename__ = 'table1_map'
    id = Column(Integer, ForeignKey('table4.id'), primary_key=True, nullable=False)
    map_id = Column(Integer, ForeignKey('table1.id'), primary_key=True, unique=True, nullable=False)

class Table2Map(Base):
    __tablename__ = 'table2_map'
    id = Column(Integer, ForeignKey('table4.id'), primary_key=True, nullable=False)
    map_id = Column(Integer, ForeignKey('table2.id'), primary_key=True, unique=True, nullable=False)
    
class Table3Map(Base):
    __tablename__ = 'table3_map'
    id = Column(Integer, ForeignKey('table4.id'), primary_key=True, nullable=False)
    map_id = Column(Integer, ForeignKey('table3.id'), primary_key=True, unique=True, nullable=False)

# 原有子查询定义
sub_query = select([Table5.__table__.c.id]).where(Table5.__table__.c.value=='somevalue')
# 新增：将子查询转为可复用的CTE
sub_query_cte = sub_query.cte('t5_ids')

subquery_1 = select([Table1.__table__.c.uuid.label("map_id"), Table1Map.__table__.c.id.label("id")]).select_from(Table1.__table__.join(Table1Map.__table__, Table1Map.__table__.c.map_id==Table1.__table__.c.id)).where(Table1Map.__table__.c.id.in_(sub_query_cte))
subquery_2 = select([Table2.__table__.c.uuid.label("map_id"), Table2Map.__table__.c.id.label("id")]).select_from(Table2.__table__.join(Table2Map.__table__, Table2Map.__table__.c.map_id==Table2.__table__.c.id)).where(Table2Map.__table__.c.id.in_(sub_query_cte))
subquery_3 = select([Table3.__table__.c.uuid.label("map_id"), Table3Map.__table__.c.id.label("id")]).select_from(Table3.__table__.join(Table3Map.__table__, Table3Map.__table__.c.map_id==Table3.__table__.c.id)).where(Table3Map.__table__.c.id.in_(sub_query_cte))

main_query = union(subquery_1, subquery_2, subquery_3)
print(main_query)

生成的SQL结构 #

调整后生成的SQL会将公共子查询提前到WITH块中，仅执行一次：

WITH t5_ids AS (
    SELECT TABLE5.ID
    FROM TABLE5
    WHERE TABLE5.VALUE = 'some_value'
)
SELECT TABLE1.UUID AS MAP_ID,
    TABLE1_MAP.ID AS ID
FROM TABLE1
JOIN TABLE1_MAP ON TABLE1_MAP.MAP_ID = TABLE1.ID
WHERE TABLE1_MAP.ID IN (SELECT id FROM t5_ids)
UNION
SELECT TABLE2.UUID AS MAP_ID,
    TABLE2_MAP.ID AS ID
FROM TABLE2
JOIN TABLE2_MAP ON TABLE2_MAP.MAP_ID = TABLE2.ID
WHERE TABLE2_MAP.ID IN (SELECT id FROM t5_ids)
UNION
SELECT TABLE3.UUID AS MAP_ID,
    TABLE3_MAP.ID AS ID
FROM TABLE3
JOIN TABLE3_MAP ON TABLE3_MAP.MAP_ID = TABLE3.ID
WHERE TABLE3_MAP.ID IN (SELECT id FROM t5_ids)

补充说明 #

如果使用SQLAlchemy 2.0版本，仅需要将select([列])的写法调整为select(列)即可，CTE的使用方式完全一致。

内容的提问来源于stack exchange，提问作者Sakthi Vel