Python/Airflow中插入bytes到Oracle RAW(16)列报错如何解决

报错原因 #

• 执行','.join(values)时，values列表中存在bytes类型的MD5计算结果，字符串拼接方法仅支持字符串类型元素，因此直接抛出TypeError
• 就算强制将bytes转为字符串，Oracle也无法直接识别Python的bytes格式写入RAW类型列，需要做格式转换

修复步骤 #

方案1：保持现有SQL拼接逻辑修改（不推荐，存在SQL注入风险） #

1. 新增bytes类型的处理分支，将bytes转为Oracle可识别的RAW格式
2. 追加MD5结果时统一走类型校验逻辑，避免直接插入bytes元素
   修改后的核心代码如下：

def get_md5(d):
    b = pickle.dumps(d)
    return hashlib.md5(b).digest()

# 抽离公共的单元格类型处理方法
def format_cell(cell):
    if isinstance(cell, str):
        return "'" + str(cell).replace("'", "''") + "'"
    elif cell is None:
        return 'NULL'
    elif isinstance(cell, float) and numpy.isnan(cell):
        return 'NULL'
    elif isinstance(cell, numpy.datetime64):
        return "'" + str(cell) + "'"
    elif isinstance(cell, datetime):
        return "to_date('" + cell.strftime('%Y-%m-%d %H:%M:%S') + "','YYYY-MM-DD HH24:MI:SS')"
    # 新增bytes类型处理逻辑
    elif isinstance(cell, bytes):
        # 转十六进制字符串后用Oracle的HEXTORAW函数转换为RAW类型
        return f"HEXTORAW('{cell.hex().upper()}')"
    else:
        return str(cell)

for row in rows:
    i += 1
    lst = []
    for cell in row:
        lst.append(format_cell(cell))
    # 追加当前时间
    current_time = "to_date('" + datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "','YYYY-MM-DD HH24:MI:SS')"
    lst.append(current_time)
    # 计算MD5后走格式化逻辑再加入列表
    s = get_md5(lst)
    lst.append(format_cell(s))
    
    values = tuple(lst)
    sql = f"INSERT /*+ APPEND */ INTO {table} {target_fields} VALUES  ({','.join(values)})"
    cur.execute(sql)

方案2：使用参数绑定（推荐，安全且性能更高） #

Oracle官方驱动（cx_Oracle/oracledb）原生支持bytes类型参数绑定到RAW列，不需要手动做格式转换，也能避免SQL注入风险，修改逻辑如下：

def get_md5(d):
    b = pickle.dumps(d)
    return hashlib.md5(b).digest()

for row in rows:
    i += 1
    params = []
    # 占位符列表，Oracle用:1, :2...作为参数占位符
    placeholders = []
    idx = 1
    for cell in row:
        params.append(cell)
        placeholders.append(f":{idx}")
        idx +=1
    # 追加当前时间
    params.append(datetime.now())
    placeholders.append(f":{idx}")
    idx +=1
    # 追加MD5 bytes
    s = get_md5(params)
    params.append(s)
    placeholders.append(f":{idx}")
    
    sql = f"INSERT /*+ APPEND */ INTO {table} {target_fields} VALUES ({','.join(placeholders)})"
    # 传入参数执行
    cur.execute(sql, params)

注意事项 #

RAW(16)列最多存储16字节数据，MD5 digest的结果正好是16字节，长度匹配不会出现截断问题。

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