使用encode/databases原子创建无校验冲突的父子关联对象

核心原理 #

你遇到的问题本质上是两个原因导致的：

1. encode/databases 基于 SQLAlchemy Core 实现，没有同步 ORM Session 的自动级联保存能力，你只在内存中关联了 Parent 和 Child 实例，没有手动执行 Child 的插入语句，所以 Child 不会写入数据库
2. 原子性可以直接依赖数据库事务实现：只要所有写入操作都在同一个事务上下文内，任意一步出错都会触发全量回滚，不会产生孤立的 Parent 脏数据

修改后的完整实现代码 #

from sqlalchemy import select

async def check_child_unique(db, child_list):
    # 1. 检查同批次提交的Child是否有重复name/slug
    child_names = [c.name for c in child_list]
    if len(set(child_names)) != len(child_names):
        return Error(
            code="DUPLICATE_CHILD_NAME_IN_REQUEST",
            message="提交的子项存在重复名称"
        )
    child_slugs = [c.slug for c in child_list]
    if len(set(child_slugs)) != len(child_slugs):
        return Error(
            code="DUPLICATE_CHILD_SLUG_IN_REQUEST",
            message="提交的子项存在重复标识"
        )
    # 2. 检查数据库中是否已有重复的Child name（因为Child的name是全局唯一）
    existing_children = await db.fetch_all(
        select(Child.__table__.c.name).where(Child.__table__.c.name.in_(child_names))
    )
    if existing_children:
        exist_names = [c.name for c in existing_children]
        return Error(
            code="CHILD_NAME_ALREADY_EXIST",
            message=f"子项名称 {','.join(exist_names)} 已存在"
        )
    return None

async def create_parent(db, data):
    values_input = data.values
    # 先校验Parent是否已存在
    parent_exists = await db.fetch_one(
        select(Parent.__table__).where(Parent.__table__.c.name == data.name)
    )
    if parent_exists:
        return Error(
            code="PARENT_ALREADY_EXIST",
            message=f"Parent with name {data.name} already exist"
        )
    # 预处理Child的slug
    for val in values_input:
        setattr(val, "slug", slugify(val.name))
    # 前置校验Child唯一性
    check_err = await check_child_unique(db, values_input)
    if check_err:
        return check_err

    # 所有校验通过后进入事务执行写入
    async with db.transaction():
        # 1. 插入Parent获取ID
        parent_input = data.__dict__
        del parent_input["values"]
        parent_id = await db.execute(
            Parent.__table__.insert().values(**parent_input)
        )
        # 2. 构造Child插入数据，补充parent_id
        child_insert_data = []
        for val in values_input:
            child_dict = val.__dict__
            # 过滤掉ORM内部字段，只保留数据库需要的字段
            child_dict.pop("_sa_instance_state", None)
            child_dict["parent_id"] = parent_id
            child_insert_data.append(child_dict)
        # 3. 批量插入Child，用RETURNING返回插入后的完整数据
        inserted_children = await db.fetch_all(
            Child.__table__.insert().values(child_insert_data).returning(
                Child.__table__.c.id,
                Child.__table__.c.name,
                Child.__table__.c.slug
            )
        )
        # 构造返回值
        response_payload = {
            **data.__dict__,
            "id": parent_id,
            "choices": [dict(c) for c in inserted_children]
        }
        return ParentPayload(**response_payload)

关键改动说明 #

• 所有唯一性校验全部前置完成后再进入事务，避免无效的事务开启，也减少事务锁定时长
• 写入操作全部包裹在async with db.transaction()上下文中，不管是插入Parent失败、还是插入Child失败，都会自动触发事务回滚，不会残留任何脏数据
• 手动批量执行Child的插入语句，补充了Parent生成的ID，解决了原来Child没有写入数据库的问题
• 插入Child时使用returning子句直接获取生成的Child ID等字段，不需要额外再查询一次数据库

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