motor 是一个适用于 MongoDB 的异步 Python 驱动，通常和 asyncio 一起使用，用于非阻塞的数据库操作。 motor 是异步库， pymongo是同步库

以下是一些常见的 操作，主要使用异步方式，并对比同样的同步方式。

0.安装

shell
pip install motor pymongo

1. 导入模块并连接 MongoDB

异步（Motor）

python
import motor.motor_asyncio
from pymongo import MongoClient  # 同步操作示例使用的库

client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017")
db = client["your_database"]
collection = db["your_collection"]

同步（pymongo）

python
client = MongoClient("mongodb://localhost:27017")
db = client["your_database"]
collection = db["your_collection"]

2. 插入文档

异步插入单个文档

python
async def insert_one():
    result = await collection.insert_one({"name": "Alice", "age": 25})
    print("Inserted ID:", result.inserted_id)

同步插入单个文档

python
result = collection.insert_one({"name": "Alice", "age": 25})
print("Inserted ID:", result.inserted_id)

异步插入多个文档

python
async def insert_many():
    result = await collection.insert_many([{"name": "Bob", "age": 30}, {"name": "Charlie", "age": 35}])
    print("Inserted IDs:", result.inserted_ids)

同步插入多个文档

python
result = collection.insert_many([{"name": "Bob", "age": 30}, {"name": "Charlie", "age": 35}])
print("Inserted IDs:", result.inserted_ids)

3. 查询文档

异步查询单个文档

python
async def find_one():
    document = await collection.find_one({"name": "Alice"})
    print("Found Document:", document)

同步查询单个文档

python
document = collection.find_one({"name": "Alice"})
print("Found Document:", document)

异步查询多个文档

python
async def find_many():
    cursor = collection.find({"age": {"$gte": 25}})
    async for document in cursor:
        print(document)

同步查询多个文档

python
for document in collection.find({"age": {"$gte": 25}}):
    print(document)

4. 更新文档

异步更新单个文档

python
async def update_one():
    result = await collection.update_one({"name": "Alice"}, {"$set": {"age": 26}})
    print("Matched count:", result.matched_count)
    print("Modified count:", result.modified_count)

同步更新单个文档

python
result = collection.update_one({"name": "Alice"}, {"$set": {"age": 26}})
print("Matched count:", result.matched_count)
print("Modified count:", result.modified_count)

异步更新多个文档

python
async def update_many():
    result = await collection.update_many({"age": {"$lt": 30}}, {"$set": {"status": "young"}})
    print("Matched count:", result.matched_count)
    print("Modified count:", result.modified_count)

同步更新多个文档

python
result = collection.update_many({"age": {"$lt": 30}}, {"$set": {"status": "young"}})
print("Matched count:", result.matched_count)
print("Modified count:", result.modified_count)

5. 删除文档

异步删除单个文档

python
async def delete_one():
    result = await collection.delete_one({"name": "Alice"})
    print("Deleted count:", result.deleted_count)

同步删除单个文档

python
result = collection.delete_one({"name": "Alice"})
print("Deleted count:", result.deleted_count)

异步删除多个文档

python
async def delete_many():
    result = await collection.delete_many({"age": {"$gte": 30}})
    print("Deleted count:", result.deleted_count)

同步删除多个文档

python
result = collection.delete_many({"age": {"$gte": 30}})
print("Deleted count:", result.deleted_count)

6. 聚合查询

异步聚合

python
async def aggregate():
    pipeline = [{"$match": {"age": {"$gte": 25}}}, {"$group": {"_id": "$age", "count": {"$sum": 1}}}]
    async for doc in collection.aggregate(pipeline):
        print(doc)

同步聚合

python
pipeline = [{"$match": {"age": {"$gte": 25}}}, {"$group": {"_id": "$age", "count": {"$sum": 1}}}]
for doc in collection.aggregate(pipeline):
    print(doc)

7. 创建索引

异步创建索引

python
async def create_index():
    index_name = await collection.create_index("name")
    print("Index created:", index_name)

同步创建索引

python
index_name = collection.create_index("name")
print("Index created:", index_name)

8. 查询数据表（集合）是否存在

MongoDB 中没有显式的 "表" 概念，集合相当于关系型数据库中的表。我们可以通过 list_collection_names() 方法检查集合是否存在。

异步（Motor）

python
async def collection_exists(db, collection_name):
    collection_names = await db.list_collection_names()
    return collection_name in collection_names

同步（pymongo）

python
collection_names = db.list_collection_names()
exists = "your_collection" in collection_names
print("Collection exists:", exists)

9. 统计数据表（集合）中的数据量

可以使用 count_documents 方法来统计符合条件的文档数。若统计整个集合的数据量，条件为空 {} 即可。

异步统计整个集合的文档数量

python
async def count_documents():
    total_count = await collection.count_documents({})
    print("Total documents:", total_count)

同步统计整个集合的文档数量

python
total_count = collection.count_documents({})
print("Total documents:", total_count)

异步统计符合特定条件的文档数量

python
async def count_filtered_documents():
    count = await collection.count_documents({"age": {"$gte": 25}})
    print("Filtered documents:", count)

同步统计符合特定条件的文档数量

python
count = collection.count_documents({"age": {"$gte": 25}})
print("Filtered documents:", count)

10. 文档查询条件

MongoDB 支持丰富的查询条件，以下是一些常见的查询条件写法：

基本查询条件

• 匹配字段等于某值：{"field": value}
• 字段大于某值：{"field": {"$gt": value}}
• 字段小于某值：{"field": {"$lt": value}}
• 字段在某范围内：{"field": {"$gte": min_value, "$lte": max_value}}
• 字段不等于某值：{"field": {"$ne": value}}

python
# 异步示例
async def find_with_conditions():
    # 查询 age 大于 25 且 name 为 "Alice" 的文档
    async for doc in collection.find({"age": {"$gt": 25}, "name": "Alice"}):
        print(doc)

查询字段存在性

• 检查字段是否存在：{"field": {"$exists": True}}
• 检查字段是否不存在：{"field": {"$exists": False}}

python
# 异步示例
async def find_with_field_existence():
    async for doc in collection.find({"address": {"$exists": True}}):
        print(doc)

正则表达式查询

可以使用 $regex 查询字符串字段的特定模式：

python
# 异步示例
async def find_with_regex():
    async for doc in collection.find({"name": {"$regex": "^A"}}):
        print(doc)  # 查询 name 字段以 "A" 开头的文档

$in 和$nin 条件查询

• $in：匹配字段在指定的值列表内
• $nin：匹配字段不在指定的值列表内

python
# 异步示例
async def find_with_in_nin():
    async for doc in collection.find({"age": {"$in": [25, 30, 35]}}):
        print(doc)  # 查询 age 为 25、30 或 35 的文档

组合查询：$and、$or

• $and：同时满足多个条件
• $or：满足其中一个条件

python
# 异步示例
async def find_with_and_or():
    async for doc in collection.find({
        "$or": [
            {"age": {"$gt": 25}},
            {"name": "Alice"}
        ]
    }):
        print(doc)  # 查询 age 大于 25 或 name 为 "Alice" 的文档

11. 示例：结合所有查询条件

python
# 异步示例，结合多种查询条件
async def complex_query():
    query = {
        "$and": [
            {"age": {"$gte": 25, "$lt": 40}},        # 年龄在 25 到 40 之间
            {"name": {"$regex": "^A"}},               # 名字以 "A" 开头
            {"address": {"$exists": True}},           # 存在 address 字段
            {"status": {"$in": ["active", "pending"]}} # status 为 "active" 或 "pending"
        ]
    }
    async for doc in collection.find(query):
        print(doc)