基于Python的SQL操作：仅取新行及插入非重复行的疑问

问题1：从持续更新的数据库获取新数据，避免重复查询 #

核心思路 #

不要仅依赖固定时间范围，而是记录每次查询的最后位置（比如自增主键ID或最新时间戳），每次只查询该位置之后的数据，确保只获取新增内容。

方案1：基于自增主键ID（最可靠） #

如果表有自增主键（如id），用主键判断最准确，避免时间重复的问题：

# 从本地存储/配置读取上次查询的最大ID（首次可设为0）
last_max_id = get_last_recorded_id()

# 查询最新数据：只取ID大于上次最大值的记录
sql = ''' SELECT * FROM table WHERE id > ? '''
cur.execute(sql, [last_max_id])
new_data = cur.fetchall()

# 若有新数据，更新记录的最大ID
if new_data:
    # 假设返回的是字典格式，若为元组则按索引取，如row[0]
    max_id = max(row['id'] for row in new_data)
    save_last_recorded_id(max_id)

方案2：基于时间戳 #

如果没有自增主键，可使用时间字段（需确保时间精度足够，避免同一时间多条数据漏取）：

# 读取上次查询的最新时间戳（首次可设为一个很早的时间）
last_query_time = get_last_query_time()

sql = ''' SELECT * FROM table WHERE time > ? '''
cur.execute(sql, [last_query_time])
new_data = cur.fetchall()

if new_data:
    max_time = max(row['time'] for row in new_data)
    save_last_query_time(max_time)

问题2：插入数据时避免重复行 #

先明确前提 #

首先要给目标表table_2设置唯一约束（单字段或多字段联合），比如假设col_1和col_2联合唯一：

-- MySQL
ALTER TABLE table_2 ADD UNIQUE KEY unique_row (col_1, col_2);

-- PostgreSQL
ALTER TABLE table_2 ADD CONSTRAINT unique_row UNIQUE (col_1, col_2);

方案1：使用数据库原生语法（推荐） #

不同数据库有更简洁的原生语法，比WHERE NOT EXISTS更高效：

MySQL：ON DUPLICATE KEY UPDATE

query = '''
INSERT INTO table_2 (col_1, col_2, col_3, col_4, col_5, col_6)
VALUES(%s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE col_1 = col_1  -- 空操作，仅跳过重复行
'''

my_data = [tuple(row) for row in data_df]
cur.executemany(query, my_data)

PostgreSQL：ON CONFLICT DO NOTHING

query = '''
INSERT INTO table_2 (col_1, col_2, col_3, col_4, col_5, col_6)
VALUES(%s, %s, %s, %s, %s, %s)
ON CONFLICT (col_1, col_2) DO NOTHING
'''

my_data = [tuple(row) for row in data_df]
cur.executemany(query, my_data)

方案2：使用WHERE NOT EXISTS（通用写法） #

如果必须用WHERE NOT EXISTS，注意语法：不能直接在VALUES后加WHERE，要改成INSERT ... SELECT ...的形式，同时注意参数占位符的数量：

# 以MySQL为例，判断重复的条件要和唯一约束对应
query = '''
INSERT INTO table_2 (col_1, col_2, col_3, col_4, col_5, col_6)
SELECT %s, %s, %s, %s, %s, %s
WHERE NOT EXISTS (
    SELECT 1 FROM table_2 
    WHERE col_1 = %s AND col_2 = %s
)
'''

# 每个数据项要重复唯一字段的值，匹配占位符
my_data = []
for row in data_df:
    # row是(col1, col2, col3, col4, col5, col6)，追加col1和col2用于判断
    my_data.append(tuple(row) + (row[0], row[1]))

cur.executemany(query, my_data)

常见报错原因 #

• 没有设置唯一约束：WHERE NOT EXISTS或原生语法无法判断重复
• 语法错误：INSERT VALUES后不能直接加WHERE，必须用SELECT子句
• 参数占位符不匹配：WHERE NOT EXISTS中需要额外的参数，要确保每个数据项的长度和占位符数量一致

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