引言：

这段代码展示了如何使用 motor（异步 MongoDB 驱动）与 MongoDB 进行连接和操作，包括查询、插入、更新和删除。通过异步编程，可以提高操作的效率，特别是在大量数据处理时

准备（mongo数据库信息）：

D盘或者其他路径 创建一个文件Database.yaml内容如下：

其中的{{database}}是动态变量，后面代码中需要那个库即可填写

mongo_config:

 -  url: mongodb://127.0.0.1:27017
    database: "{{database}}"
    collections: "{{collection}}"
    field: "{{field}}"

代码封装如下：

import yaml
import motor.motor_asyncio
import asyncio
from bson import ObjectId  # 如果需要使用 ObjectId 类型

# 创建一个类来封装 MongoDB 操作
class MongoDb:

    def __init__(self, database, collection, field=None):
        # 配置路径
        config_path = r'D:\Database.yaml'
        
        # 打开配置文件并读取其内容
        with open(config_path, 'r') as file:
            yaml_content = file.read()
        
        # 解析 YAML 配置文件
        parsed_yaml = yaml.safe_load(yaml_content)
        
        # 从配置中提取 MongoDB 连接信息
        self.url = parsed_yaml['mongo_config'][0]['url']  # MongoDB 连接URL
        self.database = database  # MongoDB数据库名
        self.collection_name = collection  # 集合名称
        self.field = field  # 表单字段名

        # 创建异步MongoClient，这个连接对象用于和 MongoDB 通信
        self.client = motor.motor_asyncio.AsyncIOMotorClient(self.url)
        
        # 连接到指定的数据库
        self.db = self.client[self.database]  
        
        # 连接到指定的集合
        self.collection = self.db[self.collection_name]  

    async def find_documents(self, field_value):
        """异步查找文档"""
        try:
            # 根据字段值查找单个文档
            document = await self.collection.find_one({self.field: field_value})
            return document
        except Exception as e:
            # 捕获异常并输出错误信息
            print(f"Error in find_documents: {e}")
            return None

    async def find_documents_in_date_range(self, query):
        """异步查询日期范围内的数据"""
        try:
            # 根据条件查询多个文档，返回一个游标对象
            cursor = self.collection.find(query)
            
            # 将游标转换为列表，最多返回 100 条数据
            results = await cursor.to_list(length=100)
            return results
        except Exception as e:
            # 捕获异常并输出错误信息
            print(f"Error in find_documents_in_date_range: {e}")
            return []

    async def insert_document(self, document):
        """异步批量插入文档"""
        try:
            # 批量插入多个文档
            result = await self.collection.insert_many(document)
            return result.inserted_ids  # 返回插入的文档ID列表
        except Exception as e:
            # 捕获异常并输出错误信息
            print(f"Error in insert_document: {e}")
            return None

    async def delete_document(self, filter):
        """异步批量删除文档"""
        try:
            # 根据过滤条件删除多个文档
            result = await self.collection.delete_many(filter)
            return result.deleted_count  # 返回删除的文档数量
        except Exception as e:
            # 捕获异常并输出错误信息
            print(f"Error in delete_document: {e}")
            return 0

    async def update_document(self, filter, update):
        """异步批量更新文档"""
        try:
            # 根据过滤条件更新多个文档
            result = await self.collection.update_many(filter, update)
            return result.modified_count  # 返回修改的文档数量
        except Exception as e:
            # 捕获异常并输出错误信息
            print(f"Error in update_document: {e}")
            return 0

    async def rollback(self):
        """回滚事务（如果使用了事务）"""
        try:
            # 使用事务回滚
            async with self.client.start_session() as session:
                session.abort_transaction()
        except Exception as e:
            # 捕获异常并输出错误信息
            print(f"Error in rollback: {e}")


# 这里是示范如何使用异步方法
async def main():
    # 示例：创建 MongoDB 操作对象
    db = MongoDb(database='your_db', collection='your_collection', field='your_field')

    # 异步查询文档，返回符合条件的第一个文档
    document = await db.find_documents('field_value')
    print(document)

    # 异步插入文档，返回插入的文档IDs
    insert_result = await db.insert_document([{'name': 'Alice'}, {'name': 'Bob'}])
    print(f"Inserted document IDs: {insert_result}")

    # 异步更新文档，更新文档的名称字段
    update_result = await db.update_document({'name': 'Alice'}, {'$set': {'name': 'Charlie'}})
    print(f"Number of documents updated: {update_result}")

    # 异步删除文档，删除名称为 Charlie 的文档
    delete_result = await db.delete_document({'name': 'Charlie'})
    print(f"Number of documents deleted: {delete_result}")


if __name__ == '__main__':
    # 运行异步主函数
    asyncio.run(main())

注意： asyncio.run(main())调用适合py3.7版本及以上

py3.6版本的话，则需要变化一下引用方法：

# 运行异步代码
if __name__ == "__main__":
    loop = asyncio.get_event_loop()  # 获取当前事件循环
    loop.run_until_complete(main())  # 运行主函数