如何使用SQLAlchemy生成适用于AWS Athena的差集子查询SQL语句

错误原因 #

• 笔误：第一个子查询中column('.timestamp')多了前置.，导致列名解析异常
• Query对象误用：你使用的ORM层Query需要绑定会话/表映射，无绑定的情况下列属性映射混乱，是触发AttributeError的核心原因。纯SQL生成场景更适合用SQLAlchemy Core的语法实现
• 逻辑缺失：缺少目标SQL要求的DISTINCT去重逻辑，以及左连接后sub_query_2.username IS NULL的过滤条件
• 表关联缺失：所有查询未指定目标表analytics_events，生成SQL时无法正确匹配表结构

正确实现方案 #

直接用SQLAlchemy Core构造查询，无需创建会话，生成的SQL可直接传给Athena执行：

from sqlalchemy import Table, Column, String, TIMESTAMP, MetaData, select, distinct, and_

# 仅定义表结构用于SQL生成，无需绑定引擎
metadata = MetaData()
analytics_events = Table(
    "analytics_events",
    metadata,
    Column("category", String),
    Column("user_category", String),
    Column("timestamp", TIMESTAMP),
    Column("name", String),
    Column("username", String)
)

# 构造第一个CTE子查询：对应name=aaa的用户集合
sub_aaa = select(distinct(analytics_events.c.username)).where(
    and_(
        analytics_events.c.category == "username",
        analytics_events.c.user_category == user_category,
        analytics_events.c.timestamp >= starting_date,
        analytics_events.c.timestamp <= ending_aaa_date,
        analytics_events.c.name == "aaa"
    )
).cte("sub_query_1")

# 构造第二个CTE子查询：对应name=bbb的用户集合
sub_bbb = select(distinct(analytics_events.c.username)).where(
    and_(
        analytics_events.c.category == "username",
        analytics_events.c.user_category == user_category,
        analytics_events.c.timestamp >= starting_date,
        analytics_events.c.timestamp <= now,
        analytics_events.c.name == "bbb"
    )
).cte("sub_query_2")

# 构造最终查询：左连接过滤仅在第一个集合存在的用户
final_query = select(sub_aaa.c.username).outerjoin(
    sub_bbb, sub_aaa.c.username == sub_bbb.c.username
).where(sub_bbb.c.username.is_(None))

# 生成可直接执行的SQL字符串
athena_sql = str(final_query.compile(compile_kwargs={"literal_binds": True}))

生成的athena_sql和你预期的SQL结构完全一致，可直接通过boto3提交到Athena执行。

内容的提问来源于stack exchange，提问作者Jérémy Litaize