如何在Mongoengine中通过索引更新嵌入列表字段的元素？

我有一个使用GraphQL和Mongoengine的项目，想要更新嵌入列表字段中的特定元素，不想从客户端获取整个列表，而是希望仅获取新值、索引以及要更新的字段名。以下是示例Post模型：

class Tag(EmbeddedDocument):
   value = StringField()
   description = StringField()

class Post(Document):
    content = StringField()
    tags = EmbbeddedDocumentListField(Tag)  # 注意：此处拼写错误，应为EmbeddedDocumentListField

class PostRepository:
    def get(self, **kwargs):
        return Post.object.filter(**kwargs).first()  # 注意：应为Post.objects
    
    def update(self, query: dict, update_fields: dict):
        post = self.get(**query)
        post.modify(query=None, **update_fields)
        return post

我的解析器代码如下：

def update_post_resolver(id: str):
    post_repository = PostRepository()
    return post_repository.update({"id": id}, {"set__tags__0": {"value": "Python", "description": "..."}})

但运行时出现如下报错：

raise ValueError(
ValueError: The source SON object needs to be of type 'dict' but a '<class 'str'>' was found

raise InvalidQueryError(
mongoengine.errors.InvalidQueryError: Querying the embedded document Tag failed, due to an invalid query value

请问是否可以在Mongoengine中使用下划线关键字参数，通过索引更新特定的Tag元素？

解决方案 #

可以通过下划线关键字参数结合索引更新嵌入列表的特定元素，但你的写法存在问题，以下是修正方案：

1. 错误原因分析 #

你直接给set__tags__0传入字典，试图替换整个Tag对象，但Mongoengine要求你明确指定要更新的内嵌字段路径，而非直接传入完整的文档字典。另外注意模型中的两处拼写错误：EmbbeddedDocumentListField应为EmbeddedDocumentListField，Post.object应为Post.objects，这些错误也可能引发潜在问题。

2. 正确的更新写法 #

要更新内嵌列表中指定索引元素的字段，需要将路径细化到具体字段，格式为set__[列表字段名]__[索引]__[内嵌字段名]：

单字段更新

{"set__tags__0__value": "Python"}

多字段更新

{
    "set__tags__0__value": "Python",
    "set__tags__0__description": "..."
}

修正后的解析器代码：

def update_post_resolver(id: str):
    post_repository = PostRepository()
    return post_repository.update(
        {"id": id}, 
        {
            "set__tags__0__value": "Python",
            "set__tags__0__description": "..."
        }
    )

3. 更高效的数据库直接更新 #

如果不需要先获取文档对象，可以直接使用update_one方法在数据库层面更新，避免额外的查询开销：

更新仓库类方法：

class PostRepository:
    def get(self, **kwargs):
        return Post.objects.filter(**kwargs).first()
    
    def update_embedded_field(self, query: dict, index: int, update_data: dict):
        update_fields = {}
        for key, value in update_data.items():
            update_fields[f"set__tags__{index}__{key}"] = value
        Post.objects(**query).update_one(**update_fields)

解析器调用示例：

def update_post_resolver(id: str):
    post_repository = PostRepository()
    post_repository.update_embedded_field(
        {"id": id},
        0,
        {"value": "Python", "description": "..."}
    )
    return post_repository.get(id=id)

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