如何从E11000 DuplicateKeyError中获取具体重复项信息？

当然可以从DuplicateKeyError里提取具体是哪个字段重复！pymongo的这个异常对象包含了足够的细节信息，我们可以解析它来区分用户名重复和邮箱重复的场景，这样就能为两种情况编写独立的测试用例了。

原理说明 #

pymongo抛出的DuplicateKeyError异常包含了详细的错误元数据，其中的keyPattern字段会指明触发唯一约束的索引名称。由于你已经为username和email分别创建了唯一索引，MongoDB默认会将索引命名为username_1和email_1，我们可以通过解析这个字段来判断具体是哪个字段重复。

修改register()函数的异常处理逻辑 #

更新你的异常捕获代码，根据不同的重复字段返回差异化的提示信息：

def register():
    # Get information about user
    username = request.get_json().get("username")
    password = request.get_json().get("password")
    name = request.get_json().get("name")
    email = request.get_json().get("email")
    # Put information about user in a tuple
    values = (
        None,
        username,
        User.hash_password(password),
        name,
        email,
        None
    )
    try:
        # Create user and update session
        User(*values).create()
        ActiveUser.logged_in = True
        ActiveUser.username = username
        info_log.info("User %s registered successfully." % username)
        return jsonify(success=True, message="Registration successful!")
    except pymongo.errors.DuplicateKeyError as e:
        # 兼容不同pymongo版本的错误信息获取方式
        error_info = e.details if hasattr(e, 'details') else e.args[0]
        key_pattern = error_info.get('keyPattern', {})
        
        if 'username' in key_pattern:
            message = "Username already exists!"
        elif 'email' in key_pattern:
            message = "Email already exists!"
        else:
            message = "Duplicated username or email!"
            
        return jsonify(success=False, message=message), 403

注：不同版本的pymongo存储错误细节的位置可能不同，这里做了兼容处理——优先读取e.details，如果不存在则从e.args[0]中获取信息。

更新单元测试用例 #

现在你可以为两种重复场景编写独立的测试，分别验证对应的错误提示：

# Register helper function
def register(self, username, password, name, email):
    return self.app.post(
        "/register",
        data = json.dumps(dict(username = username, password = password, name = name, email = email)),
        content_type='application/json',
        follow_redirects = True
    )

def test_02_valid_user_registration(self):
    response = self.register('test', '12345678', 'Tester 1', 'test@mail.mail')
    self.assertEqual(response.status_code, 200)
    self.assertIn(b'Registration successful!', response.data)

def test_03_invalid_user_registration_duplicate_username(self):
    response = self.register('test', '12345678', 'Tester 2', 'test1@mail.mail')
    self.assertEqual(response.status_code, 403)
    self.assertIn(b'Username already exists!', response.data)

def test_04_invalid_user_registration_duplicate_email(self):
    response = self.register('test2', '12345678', 'Tester 3', 'test@mail.mail')
    self.assertEqual(response.status_code, 403)
    self.assertIn(b'Email already exists!', response.data)

额外说明 #

你提到这更偏向集成测试，这一点完全没错——因为我们需要验证数据库的唯一约束是否正常生效，以及服务端对数据库异常的处理是否符合预期。在实际项目中，这类测试是很有价值的，它能确保你的业务逻辑和数据库约束之间的交互是正确的。

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