如何通过更少SQL查询实现Python对接MySQL的报表统计需求

方案1：MySQL侧一次计算所有统计值（推荐，资源消耗最低） #

你所有查询的f_id固定，仅col1-col5为100组可变组合，直接将所有组合拼入单次SQL的筛选条件，用GROUP BY直接返回所有组合的统计值，不需要拉取全量原始数据，IO开销最低。

from collections import defaultdict

# 先把所有需要查询的col1-col5组合整理成列表
query_combs = [
    ('a', 'a', 'a', 'x', 1),
    ('a', 'b', 'a', 'x', 1),
    ('a', 'a', 'b', 'x', 1),
    ('a', 'a', 'a', 'z', 1),
    ('a', 'a', 'a', 'z', 2),
    # 剩下的94组组合补在这里
]
fixed_f_id = 123

# 构造SQL
mycursor = mydb.cursor()
# 每个组合对应一个条件片段
condition_fragment = "(" + " AND ".join([f"col{i}=%s" for i in range(1,6)]) + ")"
all_conditions = " OR ".join([condition_fragment]*len(query_combs))
sql = f"""
SELECT col1, col2, col3, col4, col5, COUNT(*) as cnt
FROM mytable
WHERE f_id = %s AND ({all_conditions})
GROUP BY col1, col2, col3, col4, col5
"""
# 构造参数：第一个是f_id，后面依次展开所有组合的元素
params = [fixed_f_id]
for comb in query_combs:
    params.extend(comb)

mycursor.execute(sql, params)
# 结果转字典，方便后续查询
count_map = defaultdict(int)
for col1, col2, col3, col4, col5, cnt in mycursor.fetchall():
    count_map[(col1, col2, col3, col4, col5)] = cnt

# 使用的时候直接取即可
print(count_map[('a', 'a', 'a', 'x', 1)])
print(count_map[('a', 'b', 'a', 'x', 1)])

方案2：拉取全量数据用Python原生库统计（适合f_id对应数据量较小的场景） #

如果f_id过滤后的数据量在10万条以内，直接拉取所有col1-col5记录，用标准库collections.Counter统计即可，无额外依赖，速度极快：

from collections import Counter

fixed_f_id = 123
mycursor = mydb.cursor()
sql = "SELECT col1, col2, col3, col4, col5 FROM mytable WHERE f_id = %s"
mycursor.execute(sql, (fixed_f_id,))
all_rows = mycursor.fetchall()
# 直接统计所有组合的出现次数
count_map = Counter(all_rows)

# 使用方式和方案1完全一致
print(count_map.get(('a', 'a', 'a', 'x', 1), 0))

方案3：内存SQL查询（匹配你提到的ColdFusion特性） #

如果需要用标准SQL对内存数据做二次查询，用Python自带的标准库sqlite3即可实现，无需安装任何第三方依赖：

import sqlite3

fixed_f_id = 123
# 第一步：从MySQL拉取全量f_id对应的数据
mycursor = mydb.cursor()
sql = "SELECT col1, col2, col3, col4, col5 FROM mytable WHERE f_id = %s"
mycursor.execute(sql, (fixed_f_id,))
all_rows = mycursor.fetchall()

# 第二步：创建内存sqlite临时表，写入数据
mem_conn = sqlite3.connect(":memory:")
mem_cursor = mem_conn.cursor()
mem_cursor.execute("CREATE TABLE temp_data (col1 text, col2 text, col3 text, col4 text, col5 int)")
mem_cursor.executemany("INSERT INTO temp_data VALUES (?,?,?,?,?)", all_rows)
mem_conn.commit()

# 第三步：直接用SQL查询内存数据，和你原来的写法几乎一致
def get_mem_result(col1, col2, col3, col4, col5):
    mem_cursor.execute("""
        SELECT COUNT(*) FROM temp_data 
        WHERE col1=? AND col2=? AND col3=? AND col4=? AND col5=?
    """, (col1, col2, col3, col4, col5))
    return mem_cursor.fetchone()[0]

# 使用方式和你原来的函数完全一样
print(get_mem_result('a', 'a', 'a', 'x', 1))
print(get_mem_result('a', 'b', 'a', 'x', 1))

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