SQLAlchemy是否支持MySQL的json_table()函数？如何转换对应SQL为ORM语句？

1. 支持情况 #

SQLAlchemy 1.4及以上版本完全支持json_table()函数，可通过func.json_table或原生SQL片段的方式调用，适配MySQL 8.0+、Oracle等原生支持该函数的数据库。

2. ORM转换实现步骤 #

第一步：定义ORM模型 #

先确保已正确映射数据库表的ORM模型：

from sqlalchemy import Column, Integer, String, JSON
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class TableA(Base):
    __tablename__ = 'table_a'
    id = Column(Integer, primary_key=True)
    tech_platform = Column(String)
    prod_id = Column(String)
    biz_type = Column(String)
    report_status = Column(String)
    report_result = Column(String)

class TableB(Base):
    __tablename__ = 'table_b'
    id = Column(Integer, primary_key=True)
    aid = Column(Integer, index=True)
    category = Column(JSON)  # 对应JSON数组类型字段

第二步：构建查询 #

提供两种实现方式，可根据数据库方言适配情况选择：

方式一：使用func.json_table（推荐）

通过SQLAlchemy的函数构造器直接生成json_table逻辑：

from sqlalchemy import select, join, and_, func, distinct, bindparam, text

# 生成JSON_TABLE的别名对象
json_table_alias = func.json_table(
    TableB.category,
    text("$[*]"),  # JSON数组遍历路径
    text("columns (element varchar(50) path '$')")  # 解析后返回的列定义
).alias('j')

# 构建最终查询语句
query = select(
    json_table_alias.c.element,
    func.count(distinct(TableA.id)).label('cnt')
).select_from(
    TableA.join(TableB, TableB.aid == TableA.id)
    .join(json_table_alias, isouter=False)  # 内连接，与原SQL逻辑一致
).where(
    and_(
        TableA.tech_platform == bindparam('tech_platform'),
        TableA.prod_id == bindparam('prod_id'),
        TableA.biz_type == bindparam('biz_type'),
        TableA.report_status.like('报告%'),
        TableA.report_result == bindparam('report_result'),
        json_table_alias.c.element != ''
    )
).group_by(json_table_alias.c.element)

方式二：直接使用原生SQL片段

若遇到方言兼容问题，可直接用原生SQL片段构造json_table部分：

from sqlalchemy import select, join, and_, func, distinct, bindparam, text

# 定义JSON_TABLE的原生SQL片段并指定返回列类型
json_table_alias = text("""
    JSON_TABLE(b.category, '$[*]' columns (element varchar(50) path '$')) j
""").columns(element=String).alias('j')

# 构建查询（后续逻辑与方式一一致）
query = select(
    json_table_alias.c.element,
    func.count(distinct(TableA.id)).label('cnt')
).select_from(
    TableA.join(TableB, TableB.aid == TableA.id).join(json_table_alias)
).where(
    and_(
        TableA.tech_platform == bindparam('tech_platform'),
        TableA.prod_id == bindparam('prod_id'),
        TableA.biz_type == bindparam('biz_type'),
        TableA.report_status.like('报告%'),
        TableA.report_result == bindparam('report_result'),
        json_table_alias.c.element != ''
    )
).group_by(json_table_alias.c.element)

第三步：执行查询 #

通过Session执行查询并传入参数：

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

# 初始化数据库连接
engine = create_engine('your_database_connection_url')
Session = sessionmaker(bind=engine)
session = Session()

# 传入参数并执行查询
results = session.execute(query, {
    'tech_platform': 'your_platform_value',
    'prod_id': 'your_prod_value',
    'biz_type': 'your_biz_type',
    'report_result': 'your_result_value'
}).all()

# 遍历处理结果
for element, cnt in results:
    print(f"元素: {element}, 去重计数: {cnt}")

3. 注意事项 #

• 必须使用SQLAlchemy 1.4及以上版本，旧版本对json_table的支持有限。
• 不同数据库方言的JSON函数语法可能有差异，比如PostgreSQL常用jsonb_to_recordset替代json_table，需根据实际数据库调整。
• 模型中category字段需定义为JSON类型，确保SQLAlchemy能正确处理JSON数据的序列化与反序列化。

内容的提问来源于stack exchange，提问作者Borden Zuo