Python异步查询迁移Azure Cosmos遇阻，求替代pymongo的异步库

针对你的场景，分两种情况给出对应解决方案：

1. 基于Cosmos DB Mongo API（匹配原pymongo技术栈） #

推荐使用motor——这是pymongo官方维护的异步驱动，完全兼容asyncio，且能直接对接Azure Cosmos DB的Mongo API（因为Cosmos Mongo API兼容Mongo 3.6+协议），写法和pymongo异步风格几乎一致，迁移成本极低。

• 安装命令：

pip install motor

• 简单异步查询示例：

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def async_query_cosmos():
    # 替换为Azure Portal获取的Cosmos Mongo API连接字符串
    client = AsyncIOMotorClient("mongodb://<account-name>:<account-key>@<account-name>.mongo.cosmos.azure.com:10255/?ssl=true&replicaSet=globaldb&retrywrites=false&maxIdleTimeMS=120000&appName=@<account-name>@")
    db = client["your-database"]
    collection = db["your-collection"]

    # 执行异步查询
    async for document in collection.find({"status": "active"}):
        print(document)

asyncio.run(async_query_cosmos())

注意：连接字符串需从Azure Portal的Cosmos账户页面获取，确保启用了Mongo API模式，SSL参数配置正确。

2. 基于Cosmos DB SQL API #

如果使用Cosmos原生SQL API，可使用微软官方提供的异步SDK azure-cosmos[aio]，原生支持asyncio，完全适配Cosmos的所有特性。

• 安装命令：

pip install azure-cosmos[aio]

• 简单异步查询示例：

import asyncio
from azure.cosmos.aio import CosmosClient
from azure.cosmos import PartitionKey

async def async_query_sql_api():
    # 替换为你的Cosmos账户URL和密钥
    client = CosmosClient("https://<account-name>.documents.azure.com:443/", "<account-key>")
    database = await client.create_database_if_not_exists(id="your-database")
    container = await database.create_container_if_not_exists(
        id="your-container",
        partition_key=PartitionKey(path="/partition-key")
    )

    # 参数化查询示例
    query = "SELECT * FROM c WHERE c.category = @category"
    parameters = [{"name": "@category", "value": "electronics"}]
    
    items = container.query_items(
        query=query,
        parameters=parameters,
        enable_cross_partition_query=True
    )

    async for item in items:
        print(item)

asyncio.run(async_query_sql_api())

额外提示 #

• 使用motor对接Mongo API时，部分Mongo高级特性（如事务、某些聚合操作）可能受Cosmos兼容性限制，需验证后使用。
• 官方异步SDK的API设计和pymongo差异较大，但能完整利用Cosmos的原生功能，适合需要深度定制的SQL API场景。

内容的提问来源于stack exchange，提问作者Carabes