SQLAlchemy场景下父记录未知是否存在时添加关联子记录的方案咨询

针对你的场景，最优方案是直接使用MariaDB原生支持的INSERT ... ON DUPLICATE KEY UPDATE原子语法（SQLAlchemy 1.4及以上版本已原生适配），全程无需提前查询、无分支判断、无并发竞态问题，性能远高于先查后写的实现。

方案1：单条高频写入实现 #

from sqlalchemy.dialects.mysql import insert

Model = automap_base()
Model.prepare(sql_engine, reflect=True)
Meta = Model.classes.meta
AttrsNew = Model.classes.attrs_new

def write_single_entry(firstname, lastname, weight, height):
    with Session(sql_engine) as session:
        # 构造Meta表Upsert语句，姓名冲突时不修改原有数据，仅返回主键
        meta_upsert = insert(Meta).values(
            firstname=firstname,
            lastname=lastname
        ).on_duplicate_key_update(
            # 无意义更新仅用于触发主键返回，不会修改原有数据
            data_id=Meta.data_id
        )
        # 执行并拿到对应记录的data_id
        result = session.execute(meta_upsert)
        current_data_id = result.inserted_primary_key[0]

        # 构造属性表Upsert语句，已存在则更新属性，不存在则新增
        attr_upsert = insert(AttrsNew).values(
            data_id=current_data_id,
            weight=weight,
            height=height
        ).on_duplicate_key_update(
            weight=insert.values.weight,
            height=insert.values.height
        )
        session.execute(attr_upsert)
        session.commit()

# 调用示例
write_single_entry(firstname='John', lastname='Doe', weight=80, height=180)
write_single_entry(firstname='Amanda', lastname='Smith', weight=60, height=165)

方案2：批量写入优化实现 #

如果是批量数据接入，可以用批量Upsert减少数据库交互次数，性能比循环单条写入高10倍以上：

from sqlalchemy.dialects.mysql import insert
from sqlalchemy import select

def batch_write_entries(entry_list):
    # entry_list格式：[{"firstname":"xxx", "lastname":"xxx", "weight":xx, "height":xx}, ...]
    with Session(sql_engine) as session:
        # 批量Upsert Meta表
        meta_values = [{"firstname": e["firstname"], "lastname": e["lastname"]} for e in entry_list]
        meta_upsert = insert(Meta).values(meta_values).on_duplicate_key_update(data_id=Meta.data_id)
        session.execute(meta_upsert)

        # 批量查询所有对应记录的data_id
        name_filters = [
            (Meta.firstname == e["firstname"]) & (Meta.lastname == e["lastname"])
            for e in entry_list
        ]
        meta_records = session.execute(
            select(Meta.data_id, Meta.firstname, Meta.lastname).where(*name_filters)
        ).all()
        name_to_id = {(r.firstname, r.lastname): r.data_id for r in meta_records}

        # 批量Upsert属性表
        attr_values = [
            {"data_id": name_to_id[(e["firstname"], e["lastname"])], "weight": e["weight"], "height": e["height"]}
            for e in entry_list
        ]
        attr_upsert = insert(AttrsNew).values(attr_values).on_duplicate_key_update(
            weight=insert.values.weight,
            height=insert.values.height
        )
        session.execute(attr_upsert)
        session.commit()

注意事项 #

• 如果不需要覆盖已存在的属性数据，可以把属性表的on_duplicate_key_update逻辑去掉，换成prefix_with("IGNORE")参数实现冲突时跳过写入
• 该方案依赖MariaDB/MySQL专属语法，无需额外适配即可在你的现有环境运行，原子性由数据库保证，不会出现唯一键冲突问题

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