华为云 Python NoSql For DDS（兼容MongoDb） 演示

1.说明

本项目将演示通过pymongo对DDS进行操作。

1.1 环境准备

• python3.9.2
• Pycharm 2022+ / Visual Studio 2022+
• MongoDb 4.0+

1.2必备Pypi包

• pymongo == 4.2.0

1.3 数据库选型

• 华为云文档数据库服务 DDS（兼容 MongoDb）
  更多详情，请参考 DDS 成长地图

1.4 核心代码介绍

• src\mongodb.py

class MongoConnect:

    def __new__(cls, *args, **kwargs):
        # 获取配置文件路径
        dir_path = os.path.split(os.path.split(__file__)[0])[0]
        config_path = os.path.join(dir_path, r"config\config.json")
        with open(config_path, "r") as config_file:
            # 反序列化配置文件
            config = json.load(config_file)
        mongo_address = config.get("mongodb", None)
        if not mongo_address:
            raise Exception("连接地址不存在或为空")
        return pymongo.MongoClient(mongo_address)

• test\mongodb_test.py
  对mongo进行增删改查测试

class MongoDbTest(unittest.TestCase):
    mc = None

    @classmethod
    def setUpClass(cls) -> None:
        cls.mc = MongoConnect()

    @classmethod
    def tearDownClass(cls) -> None:
        cls.mc.close()

    def test_0_insert_value(self):
        """测试添加值"""
        print("开始添加值")
        collection = self.mc.test.test
        collection.insert_one({"id": "test"})
        if collection.find_one({"id": "test"}):
            self.assertTrue(True, "数据插入成功")
        else:
            self.assertTrue(False, "数据插入失败")

    def test_1_update_value(self):
        """测试更新值"""
        print("开始更新值")
        collection = self.mc.test.test
        collection.update_one({"id": "test"}, {"$set": {"id": "test2"}})
        if collection.find_one({"id": "test2"}):
            self.assertTrue(True, "数据更新成功")
        else:
            self.assertTrue(False, "数据更新失败")

    def test_2_delete_value(self):
        """测试删除值"""
        print("开始删除值")
        collection = self.mc.test.test
        collection.delete_many({"id": "test2"})
        if not collection.find_one({"id": "test2"}):
            self.assertTrue(True, "数据删除成功")
        else:
            self.assertTrue(False, "数据删除失败")

• config\config.json

{
  "mongodb": "mongodb://username:password@ip_address:port/?authSource=admin"
}

username:用户名
password:用户密码
ip_address:数据库IP地址
port:数据库端口