引言：
这段代码展示了如何使用 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()) # 运行主函数