从嵌套循环提取API数据插入MySQL，含关联ID查询问题

先梳理现有代码的核心问题 #

1. 示例数据中cars是单个字典，原代码for car in cars:会遍历字典键，逻辑错误
2. INSERT语句字段数与VALUES占位符数不匹配
3. get_my_system_id函数存在语法错误（if_row应为if id_row，return null应为return None）
4. 未收集所有vehicle_id到all_vehicles字段
5. 未正确调用关联ID查询函数并插入结果

问题1：处理数量不固定的vehicle_id存入all_vehicles列 #

将同一个Consist下的所有车辆ID收集为逗号分隔字符串（适合简单查询）或JSON数组（适合复杂结构），根据MySQL表字段类型选择：

• 字符串存储：用,.join提取所有车辆ID
• JSON存储：用json.dumps转成JSON格式（需确保表字段类型为JSON）

问题2：正确调用get_my_system_id函数获取关联ID #

1. 修正函数语法错误，复用数据库连接避免资源浪费
2. 在循环中获取ext_id后调用函数，将返回的my_system_id加入插入数据

完整修正代码 #

修正后的关联ID查询函数

def get_my_system_id(ext_id, db_conn):
    cursor = db_conn.cursor()    
    sql = """SELECT my_system_id FROM table WHERE ext_id = %s"""
    data = (ext_id,)
    cursor.execute(sql, data)
    id_row = cursor.fetchone()
    cursor.close()  # 及时关闭游标
    if id_row is not None: 
        return id_row[0]
    else:
        return None

主插入逻辑代码

import datetime
import json  # 用JSON存储车辆ID时需导入

# 假设已建立数据库连接db_conn
datetime_received = datetime.now().strftime('%Y-%m-%d %H:%M:%S')  # 转成MySQL兼容的时间格式
car_dealer_id = 11
ind_id = 8
dealer_name = 'XXX'

# 直接处理单个car字典（示例数据中cars不是列表）
car = cars
code = car['Code']
# 拆分API返回的ISO时间为日期和时间部分
r_date_time = datetime.datetime.fromisoformat(car['RDate'].rstrip('0').replace('+01:00', '+01:00'))
start_date = r_date_time.strftime('%Y-%m-%d')
start_time = r_date_time.strftime('%H:%M:%S')
end_date = start_date  # 保持原逻辑中end_date与start_date一致

insert_data = []

for portion in car['Consists']['Portions']:
    location = portion['Location']
    
    for consist in portion['Consist']:
        ext_id = consist['ExtId']
        # 获取关联的my_system_id
        my_system_id = get_my_system_id(ext_id, db_conn)
        
        # 方式1：收集为逗号分隔字符串
        all_vehicles = ','.join([vehicle['Id'] for vehicle in consist['Vehicles']])
        # 方式2：收集为JSON数组（需表字段为JSON类型）
        # all_vehicles = json.dumps([vehicle['Id'] for vehicle in consist['Vehicles']])
        
        # 组装单条插入数据
        row_data = (
            datetime_received,
            car_dealer_id,
            ind_id,
            dealer_name,
            code,
            start_date,
            start_time,
            end_date,
            location,
            ext_id,
            all_vehicles,
            my_system_id
        )
        insert_data.append(row_data)

# 修正后的INSERT语句，新增my_system_id字段
sql = """
INSERT INTO table (
    `datetime_received`, `car_dealer_id`, `ind_id`, `dealer_name`,
    `code`, `start_date`, `start_time`, `end_date`, `location`,
    `ext_id`, `all_vehicles`, `my_system_id`
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""

# 批量插入并处理异常
cursor = db_conn.cursor()
try:
    cursor.executemany(sql, insert_data)
    db_conn.commit()
except Exception as e:
    db_conn.rollback()
    print(f"插入失败: {str(e)}")
finally:
    cursor.close()
    db_conn.close()

关键说明 #

1. 日期处理：将API返回的ISO格式时间转换为MySQL支持的日期/时间字符串
2. 批量插入：用executemany提升插入效率，避免单条插入的性能损耗
3. 资源管理：及时关闭游标和数据库连接，避免资源泄漏
4. 安全防护：始终使用参数化查询（%s占位符），杜绝SQL注入风险

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