如何使用Python的MongoEngine在MongoDB中实现Upsert操作？

要实现「存在则更新，不存在则插入」的Upsert逻辑，你需要基于业务唯一标识字段（这里是key）执行查询，配合MongoEngine的update_one()方法并开启upsert=True参数。原来的doc.save()方式无法实现该逻辑，因为它默认以MongoDB自带的_id作为主键判断，而非你业务上的唯一键key。

正确实现代码 #

替换你原有的循环代码为以下内容：

for issue in data['issues']:
    # 整理需要更新的字段（每次执行都要同步的内容）
    update_fields = {
        'title': issue["fields"]["summary"],
        'type': issue["fields"]["issuetype"]["name"],
        'status': issue["fields"]["status"]["name"],
        'assignee': issue["fields"]["assignee"]["displayName"] if issue["fields"]["assignee"] else None,
        'labels': issue["fields"]["labels"],
        'components': [c['name'] for c in issue["fields"]["components"]],
        'storypoints': int(issue["fields"]["customfield_10002"]) if issue["fields"]["customfield_10002"] else 0,
        'sprints': [x['name'] for x in sprint_dict] if sprint_dict is not None else None,
        'updated_at': datetime.utcnow()
    }
    # 仅在插入新文档时设置的字段（比如创建时间，避免更新时覆盖）
    insert_only_fields = {
        'created': issue["fields"]["created"]
    }

    # 执行Upsert：根据key匹配，存在则更新，不存在则插入
    Issue.objects(key=issue['key']).update_one(
        set__**update_fields,
        set_on_insert__**insert_only_fields,
        upsert=True
    )

关键说明 #

1. 唯一约束保障：确保你的Issue模型中key字段设置了唯一约束，避免出现重复数据：

from mongoengine import Document, StringField, DateTimeField, ListField, IntField

class Issue(Document):
    key = StringField(unique=True, required=True)  # 必须添加唯一约束
    title = StringField()
    type = StringField()
    status = StringField()
    assignee = StringField()
    labels = ListField(StringField())
    components = ListField(StringField())
    storypoints = IntField()
    sprints = ListField(StringField())
    updated_at = DateTimeField()
    created = DateTimeField()

2. 语法解析：
   • set__**update_fields：对应MongoDB的$set操作符，用于更新匹配文档的指定字段
   • set_on_insert__**insert_only_fields：对应MongoDB的$setOnInsert操作符，仅在插入新文档时生效，不会影响已有文档的字段
   • upsert=True：核心参数，开启Upsert逻辑

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