Python实用工具：Motor——异步MongoDB操作的高效解决方案

作者：

在

一、Motor库核心概述

Motor是Python中专门用于异步操作MongoDB数据库的第三方库，它基于PyMongo开发，充分兼容asyncio异步框架，能够让开发者在异步程序中以非阻塞的方式完成MongoDB的增删改查等操作。其工作原理是将PyMongo的同步操作封装为异步协程，借助事件循环实现并发任务处理，避免同步IO操作带来的程序阻塞。

该库的优点在于：完美契合异步编程场景，提升高并发下数据库操作的效率；API设计与PyMongo高度相似，降低开发者的学习迁移成本；支持MongoDB的大部分核心功能，包括索引操作、聚合查询等。缺点则是仅适用于异步项目，同步项目中使用反而会增加复杂度；对MongoDB新版本特性的支持可能存在一定延迟。

Motor的开源协议为Apache License 2.0，这是一个对商业使用友好的开源协议，允许开发者自由修改、分发代码，且无需承担开源义务。

二、Motor库的安装步骤

在使用Motor之前，我们需要先完成库的安装，同时确保本地环境已经安装并启动了MongoDB服务，且Python版本不低于3.6（asyncio特性支持的最低版本）。

2.1 使用pip安装Motor

打开命令行终端，输入以下命令即可完成安装：

pip install motor

这条命令会从PyPI官方源下载并安装最新版本的Motor库，安装完成后，我们就可以在Python异步项目中导入并使用它。

2.2 验证安装是否成功

安装完成后，可以通过以下简单的代码片段验证Motor是否安装成功：

import motor
print(f"Motor库版本：{motor.__version__}")

运行上述代码，如果终端能够正常输出Motor的版本号，说明安装成功；若提示ModuleNotFoundError，则需要检查pip命令是否执行正确，或者Python环境是否存在冲突。

三、Motor库的核心使用方式

Motor的核心操作围绕AsyncIOMotorClient展开，这是Motor提供的异步客户端类，通过它我们可以连接MongoDB数据库、获取集合对象，并执行各类异步数据库操作。以下将详细讲解连接数据库、集合操作、数据增删改查等核心功能，并提供对应的实例代码。

3.1 连接MongoDB数据库

使用Motor连接MongoDB的方式与PyMongo类似，区别在于Motor的客户端是异步的，所有操作都需要使用await关键字。

3.1.1 基础连接示例

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def connect_to_mongodb():
    # 创建异步MongoDB客户端
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    # 验证连接是否成功
    await client.admin.command('ping')
    print("成功连接到MongoDB数据库！")
    # 指定要操作的数据库
    db = client['test_database']
    return db

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

代码说明：

首先导入asyncio和motor.motor_asyncio中的AsyncIOMotorClient类；
定义异步函数connect_to_mongodb，在函数内部创建客户端对象，传入MongoDB的连接地址（本地默认地址为mongodb://localhost:27017/）；
通过client.admin.command('ping')验证连接，该操作需要使用await关键字等待执行完成；
最后指定要操作的数据库test_database，并返回数据库对象。

3.1.2 带认证信息的连接

如果MongoDB设置了用户名和密码，连接时需要传入认证参数：

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def connect_with_auth():
    # 带用户名和密码的连接字符串格式：mongodb://用户名:密码@地址:端口/
    client = AsyncIOMotorClient('mongodb://root:123456@localhost:27017/')
    await client.admin.command('ping')
    print("带认证信息连接成功！")
    return client['test_database']

if __name__ == '__main__':
    db = asyncio.run(connect_with_auth())

代码说明：连接字符串中加入了用户名root和密码123456，适用于开启了身份验证的MongoDB环境。

3.2 集合的基本操作

在MongoDB中，集合相当于关系型数据库中的表，Motor通过db.集合名的方式获取集合对象，支持集合的创建、删除、查询存在性等操作。

3.2.1 获取集合并查询集合列表

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def collection_operations():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']

    # 获取集合对象
    collection = db['test_collection']
    print("获取集合对象成功！")

    # 查询数据库中所有的集合名称
    collection_list = await db.list_collection_names()
    print(f"数据库中的集合列表：{collection_list}")

    # 判断集合是否存在
    is_exist = 'test_collection' in collection_list
    print(f"test_collection是否存在：{is_exist}")

if __name__ == '__main__':
    asyncio.run(collection_operations())

代码说明：

通过db['test_collection']获取集合对象，也可以使用db.test_collection的方式；
db.list_collection_names()是异步方法，需要await关键字，用于获取当前数据库下的所有集合名称；
通过判断集合名是否在列表中，确认集合是否存在。

3.2.2 删除集合

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def drop_collection():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 删除集合
    await collection.drop()
    print("集合删除成功！")

    # 验证删除结果
    collection_list = await db.list_collection_names()
    print(f"删除后集合列表：{collection_list}")

if __name__ == '__main__':
    asyncio.run(drop_collection())

代码说明：调用集合对象的drop()方法可以删除指定集合，该方法为异步操作，需要await关键字。

3.3 数据的增删改查操作

数据操作是Motor的核心功能，包括插入数据、查询数据、更新数据和删除数据，所有操作均为异步协程，需要结合await关键字使用。

3.3.1 插入数据

Motor支持插入单条数据和多条数据，对应的方法分别是insert_one()和insert_many()。

插入单条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def insert_single_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 定义要插入的数据
    data = {
        'name': '张三',
        'age': 25,
        'gender': '男',
        'hobbies': ['篮球', '编程']
    }

    # 插入单条数据
    result = await collection.insert_one(data)
    print(f"插入数据的ID：{result.inserted_id}")

if __name__ == '__main__':
    asyncio.run(insert_single_data())

代码说明：

定义一个字典类型的数据，符合MongoDB的文档格式；
调用insert_one()方法插入数据，该方法返回一个InsertOneResult对象；
通过result.inserted_id可以获取插入数据的唯一ID（ObjectId）。

插入多条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def insert_multiple_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 定义多条数据
    data_list = [
        {'name': '李四', 'age': 22, 'gender': '女'},
        {'name': '王五', 'age': 28, 'gender': '男'},
        {'name': '赵六', 'age': 30, 'gender': '男'}
    ]

    # 插入多条数据
    result = await collection.insert_many(data_list)
    print(f"插入数据的ID列表：{result.inserted_ids}")

if __name__ == '__main__':
    asyncio.run(insert_multiple_data())

代码说明：

定义一个包含多个字典的列表，作为要插入的多条数据；
调用insert_many()方法插入数据，返回InsertManyResult对象；
通过result.inserted_ids获取所有插入数据的ID列表。

3.3.2 查询数据

查询数据是MongoDB的核心功能之一，Motor提供了find()和find_one()方法，分别用于查询多条数据和单条数据，支持条件过滤、字段投影、排序、分页等操作。

查询单条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def find_single_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 查询单条数据：查询name为张三的文档
    data = await collection.find_one({'name': '张三'})
    if data:
        print(f"查询到的数据：{data}")
    else:
        print("未查询到对应数据")

if __name__ == '__main__':
    asyncio.run(find_single_data())

代码说明：find_one()方法接收一个查询条件字典，返回符合条件的第一条文档，如果没有符合条件的文档，则返回None。

查询多条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def find_multiple_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 查询多条数据：查询age大于25的文档
    cursor = collection.find({'age': {'$gt': 25}})
    # 遍历游标获取数据
    async for data in cursor:
        print(f"查询到的数据：{data}")

if __name__ == '__main__':
    asyncio.run(find_multiple_data())

代码说明：

find()方法接收查询条件字典，返回一个异步游标对象（AsyncIOMotorCursor）；
使用async for循环遍历游标，获取所有符合条件的文档；
查询条件中使用了MongoDB的查询操作符$gt（大于），类似的还有$lt（小于）、$eq（等于）等。

条件过滤、排序与分页

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def find_data_with_filter():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 1. 条件过滤：查询gender为男，且age在20-30之间的文档
    query = {
        'gender': '男',
        'age': {'$gte': 20, '$lte': 30}
    }
    # 2. 字段投影：只返回name、age字段，不返回_id字段
    projection = {'_id': 0, 'name': 1, 'age': 1}
    # 3. 排序：按age降序排列
    sort = [('age', -1)]
    # 4. 分页：跳过前1条数据，获取2条数据
    skip = 1
    limit = 2

    cursor = collection.find(query, projection).sort(sort).skip(skip).limit(limit)
    async for data in cursor:
        print(f"过滤后的数据：{data}")

if __name__ == '__main__':
    asyncio.run(find_data_with_filter())

代码说明：

query字典定义查询条件，使用$gte（大于等于）和$lte（小于等于）操作符限定age范围；
projection字典定义返回的字段，1表示返回，0表示不返回；
sort()方法接收排序规则列表，-1表示降序，1表示升序；
skip()方法用于跳过指定数量的文档，limit()方法用于限制返回的文档数量，实现分页功能。

3.3.3 更新数据

Motor支持更新单条数据和多条数据，对应的方法是update_one()和update_many()，更新操作需要使用MongoDB的更新操作符。

更新单条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def update_single_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 查询条件：name为张三
    query = {'name': '张三'}
    # 更新内容：将age增加1，添加city字段
    update = {
        '$inc': {'age': 1},
        '$set': {'city': '北京'}
    }

    result = await collection.update_one(query, update)
    print(f"匹配的文档数量：{result.matched_count}")
    print(f"修改的文档数量：{result.modified_count}")

if __name__ == '__main__':
    asyncio.run(update_single_data())

代码说明：

query字典定义要更新的文档条件；
update字典使用更新操作符$inc（增加数值）和$set（设置字段值）定义更新内容；
update_one()方法只更新符合条件的第一条文档，返回UpdateResult对象，通过matched_count和modified_count查看匹配和修改的文档数量。

更新多条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def update_multiple_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 查询条件：gender为男
    query = {'gender': '男'}
    # 更新内容：设置city为上海
    update = {'$set': {'city': '上海'}}

    result = await collection.update_many(query, update)
    print(f"匹配的文档数量：{result.matched_count}")
    print(f"修改的文档数量：{result.modified_count}")

if __name__ == '__main__':
    asyncio.run(update_multiple_data())

代码说明：update_many()方法会更新所有符合条件的文档，适用于批量更新场景。

3.3.4 删除数据

删除数据的方法包括delete_one()和delete_many()，分别用于删除单条和多条符合条件的文档。

删除单条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def delete_single_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 查询条件：name为赵六
    query = {'name': '赵六'}
    result = await collection.delete_one(query)
    print(f"删除的文档数量：{result.deleted_count}")

if __name__ == '__main__':
    asyncio.run(delete_single_data())

代码说明：delete_one()方法删除符合条件的第一条文档，返回DeleteResult对象，通过deleted_count查看删除的文档数量。

删除多条数据

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def delete_multiple_data():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['test_database']
    collection = db['test_collection']

    # 查询条件：age小于25
    query = {'age': {'$lt': 25}}
    result = await collection.delete_many(query)
    print(f"删除的文档数量：{result.deleted_count}")

if __name__ == '__main__':
    asyncio.run(delete_multiple_data())

代码说明：delete_many()方法删除所有符合条件的文档，适用于批量删除场景，使用时需要谨慎，避免误删数据。

四、Motor库的实际应用案例

下面我们结合一个异步Web服务的场景，展示Motor库的实际应用。我们将使用FastAPI框架搭建一个简单的用户信息管理接口，实现用户信息的增删改查，所有数据库操作均通过Motor完成。

4.1 环境准备

首先安装FastAPI和Uvicorn（ASGI服务器，用于运行FastAPI应用）：

pip install fastapi uvicorn

4.2 编写接口代码

from fastapi import FastAPI, HTTPException
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel
import asyncio

# 定义FastAPI应用
app = FastAPI(title="用户信息管理接口", version="1.0")

# 定义数据模型（请求体）
class UserModel(BaseModel):
    name: str
    age: int
    gender: str
    city: str = None

# 全局数据库连接
client = AsyncIOMotorClient('mongodb://localhost:27017/')
db = client['user_db']
collection = db['user_collection']

# 1. 创建用户接口（POST）
@app.post("/users/", summary="创建新用户")
async def create_user(user: UserModel):
    user_dict = user.dict()
    result = await collection.insert_one(user_dict)
    return {"message": "用户创建成功", "user_id": str(result.inserted_id)}

# 2. 查询单个用户接口（GET）
@app.get("/users/{user_name}", summary="根据用户名查询用户")
async def get_user(user_name: str):
    user = await collection.find_one({"name": user_name}, {"_id": 0})
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

# 3. 查询所有用户接口（GET）
@app.get("/users/", summary="查询所有用户")
async def get_all_users(skip: int = 0, limit: int = 10):
    users = []
    cursor = collection.find({}, {"_id": 0}).skip(skip).limit(limit)
    async for user in cursor:
        users.append(user)
    return {"total": len(users), "users": users}

# 4. 更新用户接口（PUT）
@app.put("/users/{user_name}", summary="更新用户信息")
async def update_user(user_name: str, user: UserModel):
    update_data = user.dict(exclude_unset=True)
    result = await collection.update_one(
        {"name": user_name},
        {"$set": update_data}
    )
    if result.matched_count == 0:
        raise HTTPException(status_code=404, detail="用户不存在")
    return {"message": "用户信息更新成功"}

# 5. 删除用户接口（DELETE）
@app.delete("/users/{user_name}", summary="删除用户")
async def delete_user(user_name: str):
    result = await collection.delete_one({"name": user_name})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="用户不存在")
    return {"message": "用户删除成功"}

if __name__ == '__main__':
    import uvicorn
    # 运行FastAPI应用
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.3 代码说明与运行测试

代码说明
- 首先导入FastAPI、Motor等相关模块，定义UserModel作为请求体的数据模型；
- 创建全局的Motor客户端和集合对象，确保整个应用共享一个数据库连接；
- 实现5个核心接口：创建用户、查询单个用户、查询所有用户、更新用户、删除用户，所有接口均为异步函数，数据库操作使用await关键字；
- 使用HTTPException处理异常情况，如用户不存在时返回404状态码。
运行测试
- 运行上述代码，启动Uvicorn服务器；
- 打开浏览器访问http://localhost:8000/docs，可以看到FastAPI自动生成的接口文档；
- 在文档页面中可以直接测试各个接口，例如点击/users/的POST接口，输入用户信息后执行，即可在MongoDB中插入一条用户数据。

五、Motor库相关资源

PyPI地址：https://pypi.org/project/Motor
Github地址：https://github.com/mongodb/motor
官方文档地址：https://motor.readthedocs.io/

关注我，每天分享一个实用的Python自动化工具。

实用工具

Python实用工具：Motor——异步MongoDB操作的高效解决方案

一、Motor库核心概述

二、Motor库的安装步骤

2.1 使用pip安装Motor

2.2 验证安装是否成功

三、Motor库的核心使用方式

3.1 连接MongoDB数据库

3.1.1 基础连接示例

3.1.2 带认证信息的连接

3.2 集合的基本操作

3.2.1 获取集合并查询集合列表

3.2.2 删除集合

3.3 数据的增删改查操作

3.3.1 插入数据

插入单条数据

插入多条数据

3.3.2 查询数据

查询单条数据

查询多条数据

条件过滤、排序与分页

3.3.3 更新数据

更新单条数据

更新多条数据

3.3.4 删除数据

删除单条数据

删除多条数据

四、Motor库的实际应用案例

4.1 环境准备

4.2 编写接口代码

4.3 代码说明与运行测试

五、Motor库相关资源

更多文章

Python实用工具Beam：轻量化任务调度与异步执行入门教程

Python实用工具：Celery分布式任务队列入门与实战教程

Python实用工具：Apache Airflow 从入门到实战保姆级教程

Python Squirrel库入门教程：高效数据缓存与持久化工具

Python实用工具：Motor——异步MongoDB操作的高效解决方案

一、Motor库核心概述

二、Motor库的安装步骤

2.1 使用pip安装Motor

2.2 验证安装是否成功

三、Motor库的核心使用方式

3.1 连接MongoDB数据库

3.1.1 基础连接示例

3.1.2 带认证信息的连接

3.2 集合的基本操作

3.2.1 获取集合并查询集合列表

3.2.2 删除集合

3.3 数据的增删改查操作

3.3.1 插入数据

插入单条数据

插入多条数据

3.3.2 查询数据

查询单条数据

查询多条数据

条件过滤、排序与分页

3.3.3 更新数据

更新单条数据

更新多条数据

3.3.4 删除数据

删除单条数据

删除多条数据

四、Motor库的实际应用案例

4.1 环境准备

4.2 编写接口代码

4.3 代码说明与运行测试

五、Motor库相关资源

更多文章

Python实用工具Beam：轻量化任务调度与异步执行入门教程

Python实用工具：Celery分布式任务队列入门与实战教程

Python实用工具：Apache Airflow 从入门到实战 保姆级教程

Python Squirrel库入门教程：高效数据缓存与持久化工具

Python实用工具：Apache Airflow 从入门到实战保姆级教程