Context

I'm currently working on a project using the new PyMongo asynchronous driver, and during application startup I need to verify the database connection before proceeding. The issue is that my database constructor is synchronous, meaning I cannot use await while initializing the database connection.

In earlier versions using Motor, this was straightforward — I could simply call:

client.delegate.admin.command("ping")

which leveraged the underlying synchronous PyMongo client internally. This approach worked perfectly for me and allowed me to check connectivity before the application fully initialized. However, with Motor now being deprecated, I began migrating my code to the new PyMongo async driver — and encountered several issues.

First, the new asynchronous model in PyMongo is not as straightforward or well-organized as Motor. In Motor, commonly used classes like AsyncIOMotorClient, AsyncIOMotorDatabase, and AsyncIOMotorCollection were all easily importable from motor.motor_asyncio. In the new driver, imports are scattered and less discoverable — I had to dig deep through the package to locate them.

Second, I needed a synchronous way to verify the connection (e.g., by pinging the server). The documentation snippet https://github.com/mongodb/mongo-python-driver/blob/fd025503491e2bc939d8272770623c47dbaa3fd7/pymongo/asynchronous/mongo_client.py#L267 provided hope:

from pymongo.errors import ConnectionFailure

client = AsyncMongoClient()
try:
    # The ping command is cheap and does not require auth.
    client.admin.command('ping')
except ConnectionFailure:
    print("Server not available")

However, this example is misleading — calling client.admin.command('ping') without await does not actually work, since it must be awaited in an asynchronous context.

I also attempted to work around this limitation using a custom blocking helper:

class BlockingAsync:
    _executor = ThreadPoolExecutor(max_workers=4)
    class Wrapper:
        def _init_(self, fut): self.fut = fut

    @classmethod
    def run(cls, coro):
        future = cls._executor.submit(cls._run_in_thread, coro)
        result = future.result()
        if isinstance(result, cls.Wrapper):
            return result.fut.result()

    @staticmethod
    def _run_in_thread(coro):
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            try:
                return loop.run_until_complete(coro)
            finally:
                loop.run_until_complete(loop.shutdown_asyncgens())
                loop.close()
        else:
            if loop.is_running():
                return BlockingAsync.Wrapper(asyncio.run_coroutine_threadsafe(coro, loop))
            else:
                return loop.run_until_complete(coro)

While this allowed me to run asynchronous coroutines in a synchronous way, but the AsyncMongoClient binds itself to the temp event loop on which it was ran first. As a result, the client became unusable in the main event loop, leading to runtime errors.

In short, I'd like to request that the PyMongo async driver provide a synchronous way to verify connectivity (for example, a simple client.ping_sync() or a safe blocking wrapper). This functionality is essential for initialization and startup routines where async context is not yet available. or them to clearly mention there are not ant ways to verify database connection synchronouly in the async driver.

Definition of done

Pitfalls

What should the implementer watch out for? What are the risks?