如何在Python中将整数列表转为分块的SQL Insert单元素元组语句

问题场景 #

• 有一个规模超100万的整数列表，示例：li: list[int] = [0,1,2,3,4,5,6]
• 需要按用户指定的chunk_size分块，生成符合SQL语法的Insert语句迭代对象，期望格式：

queries: list[str] = [
    'insert into test_table (test_col) values (0),   (1),   (2),   (3),  (4)...   (x)',
    ...
    'insert into test_table (test_col) values (x+1), (x+2), (x+3), (x+4) (x+5)... (n)'
]

当前代码问题 #

使用itertools实现分块的代码如下，但生成的语句中，单元素元组格式（如(0,)）无法被SQL引擎识别，多元素元组的格式也不符合SQL要求：

import itertools

li = [0,1,2,3,4,5,6,7,8,9,10]

def grouper(n, iterable) -> tuple:
    it: iter = iter(iterable)
    while True:
        chunk: tuple = tuple(itertools.islice(it, n))
        if not chunk:
            return
        yield chunk

queries = [f"insert into testing (testing_col) values {i};" for i in grouper(2,li)]
queries[0]

解决方案 #

问题核心是直接将元组转为字符串时，格式不符合SQL的values子句要求。需要将分块后的每个元素单独包装为(val)格式，再用逗号连接。

改进代码（列表形式） #

import itertools

li = [0,1,2,3,4,5,6,7,8,9,10]

def grouper(n, iterable):
    it = iter(iterable)
    while True:
        chunk = tuple(itertools.islice(it, n))
        if not chunk:
            return
        yield chunk

queries = []
for chunk in grouper(2, li):
    # 逐个包装元素为(val)，再拼接成符合SQL的格式
    values_part = ", ".join(f"({val})" for val in chunk)
    insert_query = f"insert into testing (testing_col) values {values_part};"
    queries.append(insert_query)

# 验证第一个语句
print(queries[0])  # 输出: insert into testing (testing_col) values (0), (1);

内存友好版（生成器形式） #

针对百万级数据，使用生成器表达式可以避免一次性加载所有语句到内存：

import itertools

li = [0,1,2,3,4,5,6,7,8,9,10]

def grouper(n, iterable):
    it = iter(iterable)
    while True:
        chunk = tuple(itertools.islice(it, n))
        if not chunk:
            return
        yield chunk

# 生成器迭代对象，按需生成语句
queries = (
    f"insert into testing (testing_col) values {', '.join(f'({val})' for val in chunk)};"
    for chunk in grouper(2, li)
)

# 遍历生成器获取语句
for q in queries:
    print(q)

改进后的语句完全符合SQL语法，无论分块大小如何，每个值都会被正确包装，SQL引擎可直接识别。

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