检查 MongoDB 数据库实例是否为副本集

Question

我正在为使用事务的函数编写测试。

Transactions only work on replica sets or mongos:

pymongo.errors.OperationFailure: Transaction numbers are only allowed on a replica set member or mongos, full error: {'ok': 0.0, 'errmsg': 'Transaction numbers are only allowed on a replica set member or mongos', 'code': 20, 'codeName': 'IllegalOperation'}

如果数据库是独立的 mongod 实例，我希望跳过这些测试。

如何从 Python 代码（例如使用 pymongo 函数）检查我使用的环境是否合适？

我可以将 CI 管道配置为使用副本集。我可以并且确实在我的机器上配置了它。但我不希望这几个测试成为潜在贡献者的麻烦，所以我宁愿跳过它们。

或者，我可以通过匹配字符串来捕获 OperationFailure 并尝试确保它正是上面的错误，而不是另一个 OperationFailure，但这将是一个遗憾的解决方法。

也许我可以使用类似

的东西

MongoClient('localhost', 27017).admin.command("replSetGetStatus")

在进行艰难的试错之前，我很乐意得到明智的建议。

Answer 1

这是我想出的：

def is_replica_set():
    try:
        pymongo.MongoClient().admin.command("replSetGetStatus")
        return True
    except pymongo.errors.OperationFailure:
        return False


@pytest.mark.skipif(not is_replica_set(), reason="Replica set required")
def test_transaction():
    ...

Answer 2

发送{ismaster: 1}命令并按照规则here解析响应。

副本集节点将在响应中设置 setName 字段。

Answer 3

这是一种检查 MongoDB 实例是否为副本集的方法：

每个mongod实例都有自己的local数据库（local是数据库名称），它存储复制过程中使用的数据，以及其他实例特定的数据。 local 数据库中的 system.replset 集合将副本集的配置对象作为其单个文档保存。

要查看此配置，请从 mongo shell 发出 rs.conf()。你也可以直接查询这个集合（见下面方法中使用的代码）。

client = pymongo.MongoClient(url)
collection = client.local.system.replset
try:
    doc = collection.find()[0]
    print(doc['_id'])
    return True
except Exception as e:
    print("local.system.replset doesn't exist: " + str(e))
    return False

请注意，如果游标为空，即当 local 数据库中不存在 system.replset 集合并且查询失败（查询失败独立实例）。

在代码中，doc['_id'] 是副本集名称。

检查 MongoDB 数据库实例是否为副本集

Check MongoDB database instance is a replica set

python

mongodb

pymongo