SQLAlchemy插入一对多关系数据报错unhashable type: 'list' 求助

问题原因 #

1. SQLAlchemy Core与ORM的不兼容使用：你使用了SQLAlchemy Core的insert()方法直接对ORM模型的relationship字段（ozon_orders_info，列表类型）赋值。Core的插入操作是面向数据库表的，仅处理表的物理列，无法解析ORM层面的关联对象列表，因此抛出unhashable type: 'list'错误。
2. 低效的数据匹配与提交方式：双重循环逐个匹配数据，且每条插入后立即提交，对于超1000条数据来说，会产生大量数据库请求，性能极差，还可能引发连接超时等问题。

解决方案 #

方案一：使用ORM Session处理关联对象（简洁易维护） #

利用SQLAlchemy ORM的Session来管理对象关系，它会自动处理外键关联和cascade逻辑，代码更简洁：

from sqlalchemy.orm import Session

# 提前按articul分组订单数据，避免双重循环的低效匹配
orders_by_articul = {}
for _, order_row in data_ozon_orders.iterrows():
    articul = order_row['article']
    orders_by_articul.setdefault(articul, []).append(order_row)

with Session(engine_dwh_staging) as session:
    product_price_list = []
    for _, price_row in data_ozon_price.iterrows():
        articul = price_row['articul']
        # 创建主表对象
        price_obj = OzonProductPrice(
            articul=price_row['articul'],
            brand=price_row['brand'],
            product_price=price_row['product_price'],
            product_price_plus_invest=price_row['product_price_plus_invest'],
            date_update_db=price_row['date_update_db']
        )
        # 关联对应的订单对象
        if articul in orders_by_articul:
            for order_row in orders_by_articul[articul]:
                order_obj = OzonOrders(
                    articul=order_row['article'],
                    brand=order_row['brand'],
                    order_number=order_row['order_number'],
                    posting_number=order_row['posting_number'],
                    date_created_at=order_row['date_created_at'],
                    in_process_at=order_row['in_process_at'],
                    status=order_row['status'],
                    product_price_order=order_row['product_price_order'],
                    currency_code=order_row['currency_code'],
                    product_name=order_row['product_name'],
                    product_sku=order_row['product_sku'],
                    product_quantity=order_row['product_quantity'],
                    warehouse_name=order_row['warehouse_name'],
                    commission_percent=order_row['commission_percent'],
                    date_update_db=order_row['date_update_db']
                )
                price_obj.ozon_orders_info.append(order_obj)
        product_price_list.append(price_obj)
    
    # 批量添加所有对象到会话，统一提交
    session.add_all(product_price_list)
    session.commit()

方案二：Core批量插入（高性能，适合大数据量） #

如果数据量极大（超1000条），推荐使用Core的批量插入，绕过ORM对象实例化的开销，性能更高：

from sqlalchemy import insert

# 1. 准备主表插入数据
price_insert_data = [
    {
        'articul': row['articul'],
        'brand': row['brand'],
        'product_price': row['product_price'],
        'product_price_plus_invest': row['product_price_plus_invest'],
        'date_update_db': row['date_update_db']
    }
    for _, row in data_ozon_price.iterrows()
]

with engine_dwh_staging.connect() as conn:
    # 插入主表并返回articul对应的主键id
    insert_price_stmt = insert(OzonProductPrice).values(price_insert_data).returning(OzonProductPrice.id, OzonProductPrice.articul)
    price_results = conn.execute(insert_price_stmt)
    articul_to_price_id = {row.articul: row.id for row in price_results}
    
    # 2. 准备从表插入数据，匹配主表id
    orders_insert_data = []
    for _, order_row in data_ozon_orders.iterrows():
        articul = order_row['article']
        if articul in articul_to_price_id:
            orders_insert_data.append({
                'articul': order_row['article'],
                'brand': order_row['brand'],
                'order_number': order_row['order_number'],
                'posting_number': order_row['posting_number'],
                'date_created_at': order_row['date_created_at'],
                'in_process_at': order_row['in_process_at'],
                'status': order_row['status'],
                'product_price_order': order_row['product_price_order'],
                'currency_code': order_row['currency_code'],
                'product_name': order_row['product_name'],
                'product_sku': order_row['product_sku'],
                'product_quantity': order_row['product_quantity'],
                'warehouse_name': order_row['warehouse_name'],
                'commission_percent': order_row['commission_percent'],
                'date_update_db': order_row['date_update_db'],
                'ozon_product_price_id': articul_to_price_id[articul]
            })
    
    # 3. 批量插入从表
    if orders_insert_data:
        insert_orders_stmt = insert(OzonOrders).values(orders_insert_data)
        conn.execute(insert_orders_stmt)
    
    conn.commit()

方案选择 #

• 若数据量适中，需要利用ORM的cascade、关系校验等特性，选方案一。
• 若数据量超大（数千条以上），追求极致性能，选方案二。

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