Tutorial: Using Motor With asyncio

A guide to using MongoDB and asyncio with Motor.

Tutorial Prerequisites

You can learn about MongoDB with the MongoDB Tutorial before you learn Motor.

Using Python 3.4 or later, do:

$ python3 -m pip install motor

This tutorial assumes that a MongoDB instance is running on the default host and port. Assuming you have downloaded and installed MongoDB, you can start it like so:

$ mongod

Object Hierarchy

Motor, like PyMongo, represents data with a 4-level object hierarchy:

  • AsyncIOMotorClient represents a mongod process, or a cluster of them. You explicitly create one of these client objects, connect it to a running mongod or mongods, and use it for the lifetime of your application.
  • AsyncIOMotorDatabase: Each mongod has a set of databases (distinct sets of data files on disk). You can get a reference to a database from a client.
  • AsyncIOMotorCollection: A database has a set of collections, which contain documents; you get a reference to a collection from a database.
  • AsyncIOMotorCursor: Executing find() on an AsyncIOMotorCollection gets an AsyncIOMotorCursor, which represents the set of documents matching a query.

Creating a Client

You typically create a single instance of AsyncIOMotorClient at the time your application starts up.

>>> import motor.motor_asyncio
>>> client = motor.motor_asyncio.AsyncIOMotorClient()

This connects to a mongod listening on the default host and port. You can specify the host and port like:

>>> client = motor.motor_asyncio.AsyncIOMotorClient('localhost', 27017)

Motor also supports connection URIs:

>>> client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')

Connect to a replica set like:

>>> client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://host1,host2/?replicaSet=my-replicaset-name')

Getting a Database

A single instance of MongoDB can support multiple independent databases. From an open client, you can get a reference to a particular database with dot-notation or bracket-notation:

>>> db = client.test_database
>>> db = client['test_database']

Creating a reference to a database does no I/O and does not require an await expression.

Getting a Collection

A collection is a group of documents stored in MongoDB, and can be thought of as roughly the equivalent of a table in a relational database. Getting a collection in Motor works the same as getting a database:

>>> collection = db.test_collection
>>> collection = db['test_collection']

Just like getting a reference to a database, getting a reference to a collection does no I/O and doesn’t require an await expression.

Inserting a Document

As in PyMongo, Motor represents MongoDB documents with Python dictionaries. To store a document in MongoDB, call insert_one() in an await expression:

>>> async def do_insert():
...     document = {'key': 'value'}
...     result = await db.test_collection.insert_one(document)
...     print('result %s' % repr(result.inserted_id))
...
>>>
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_insert())
result ObjectId('...')

See general MongoDB documentation

insert

Using native coroutines

Starting in Python 3.5, you can define a native coroutine with async def instead of the coroutine decorator. Within a native coroutine, wait for an async operation with await instead of yield:

>>> async def do_insert():
...     for i in range(2000):
...         result = await db.test_collection.insert_one({'i': i})
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_insert())

Within a native coroutine, the syntax to use Motor with Tornado or asyncio is often identical.

Getting a Single Document With find_one

Use find_one() to get the first document that matches a query. For example, to get a document where the value for key “i” is less than 1:

>>> async def do_find_one():
...     document = await db.test_collection.find_one({'i': {'$lt': 1}})
...     pprint.pprint(document)
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_find_one())
{'_id': ObjectId('...'), 'i': 0}

The result is a dictionary matching the one that we inserted previously.

Note

The returned document contains an "_id", which was automatically added on insert.

See general MongoDB documentation

find

Querying for More Than One Document

Use find() to query for a set of documents. find() does no I/O and does not require an await expression. It merely creates an AsyncIOMotorCursor instance. The query is actually executed on the server when you call to_list() or execute an async for loop.

To find all documents with “i” less than 5:

>>> async def do_find():
...     cursor = db.test_collection.find({'i': {'$lt': 5}}).sort('i')
...     for document in await cursor.to_list(length=100):
...         pprint.pprint(document)
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_find())
{'_id': ObjectId('...'), 'i': 0}
{'_id': ObjectId('...'), 'i': 1}
{'_id': ObjectId('...'), 'i': 2}
{'_id': ObjectId('...'), 'i': 3}
{'_id': ObjectId('...'), 'i': 4}

A length argument is required when you call to_list to prevent Motor from buffering an unlimited number of documents.

async for

You can handle one document at a time in an async for loop:

>>> async def do_find():
...     c = db.test_collection
...     async for document in c.find({'i': {'$lt': 2}}):
...         pprint.pprint(document)
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_find())
{'_id': ObjectId('...'), 'i': 0}
{'_id': ObjectId('...'), 'i': 1}

You can apply a sort, limit, or skip to a query before you begin iterating:

>>> async def do_find():
...     cursor = db.test_collection.find({'i': {'$lt': 5}})
...     # Modify the query before iterating
...     cursor.sort('i', -1).limit(2).skip(2)
...     async for document in cursor:
...         pprint.pprint(document)
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_find())
{'_id': ObjectId('...'), 'i': 2}
{'_id': ObjectId('...'), 'i': 1}

The cursor does not actually retrieve each document from the server individually; it gets documents efficiently in large batches.

Iteration in Python 3.3 and 3.4

In Python versions without async for, handle one document at a time with fetch_next and next_object():

>>> @coroutine
... def do_find():
...     cursor = db.test_collection.find({'i': {'$lt': 5}})
...     while (yield from cursor.fetch_next):
...         document = cursor.next_object()
...         pprint.pprint(document)
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_find())
{'_id': ObjectId('...'), 'i': 0}
{'_id': ObjectId('...'), 'i': 1}
{'_id': ObjectId('...'), 'i': 2}
{'_id': ObjectId('...'), 'i': 3}
{'_id': ObjectId('...'), 'i': 4}

Counting Documents

Use count() to determine the number of documents in a collection, or the number of documents that match a query:

>>> async def do_count():
...     n = await db.test_collection.find().count()
...     print('%s documents in collection' % n)
...     n = await db.test_collection.find({'i': {'$gt': 1000}}).count()
...     print('%s documents where i > 1000' % n)
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_count())
2000 documents in collection
999 documents where i > 1000

count() uses the count command internally; we’ll cover commands below.

See also

Count command

Updating Documents

replace_one() changes a document. It requires two parameters: a query that specifies which document to replace, and a replacement document. The query follows the same syntax as for find() or find_one(). To replace a document:

>>> async def do_replace():
...     coll = db.test_collection
...     old_document = await coll.find_one({'i': 50})
...     print('found document: %s' % pprint.pformat(old_document))
...     _id = old_document['_id']
...     result = await coll.replace_one({'_id': _id}, {'key': 'value'})
...     print('replaced %s document' % result.modified_count)
...     new_document = await coll.find_one({'_id': _id})
...     print('document is now %s' % pprint.pformat(new_document))
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_replace())
found document: {'_id': ObjectId('...'), 'i': 50}
replaced 1 document
document is now {'_id': ObjectId('...'), 'key': 'value'}

You can see that replace_one() replaced everything in the old document except its _id with the new document.

Use update_one() with MongoDB’s modifier operators to update part of a document and leave the rest intact. We’ll find the document whose “i” is 51 and use the $set operator to set “key” to “value”:

>>> async def do_update():
...     coll = db.test_collection
...     result = await coll.update_one({'i': 51}, {'$set': {'key': 'value'}})
...     print('updated %s document' % result.modified_count)
...     new_document = await coll.find_one({'i': 51})
...     print('document is now %s' % pprint.pformat(new_document))
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_update())
updated 1 document
document is now {'_id': ObjectId('...'), 'i': 51, 'key': 'value'}

“key” is set to “value” and “i” is still 51.

update_one() only affects the first document it finds, you can update all of them with update_many():

await coll.update_many({'i': {'$gt': 100}},
                       {'$set': {'key': 'value'}})

See general MongoDB documentation

update

Deleting Documents

delete_many() takes a query with the same syntax as find(). delete_many() immediately removes all matching documents.

>>> async def do_delete_many():
...     coll = db.test_collection
...     n = await coll.count()
...     print('%s documents before calling delete_many()' % n)
...     result = await db.test_collection.delete_many({'i': {'$gte': 1000}})
...     print('%s documents after' % (await coll.count()))
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(do_delete_many())
2000 documents before calling delete_many()
1000 documents after

See general MongoDB documentation

remove

Commands

Besides the “CRUD” operations–insert, update, delete, and find–all other operations on MongoDB are commands. Run them using the command() method on AsyncIOMotorDatabase:

>>> from bson import SON
>>> async def use_count_command():
...     response = await db.command(SON([("count", "test_collection")]))
...     print('response: %s' % pprint.pformat(response))
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(use_count_command())
response: {'n': 1000, 'ok': 1.0, ...}

Since the order of command parameters matters, don’t use a Python dict to pass the command’s parameters. Instead, make a habit of using bson.SON, from the bson module included with PyMongo:

await db.command(SON([("distinct", "test_collection"), ("key", "my_key"]))

Many commands have special helper methods, such as create_collection() or aggregate(), but these are just conveniences atop the basic command() method.

See general MongoDB documentation

commands

A Web Application With aiohttp

Let us create a web application using aiohttp, a popular HTTP package for asyncio. Install it with:

python3 -m pip install aiohttp

We are going to make a trivial web site with two pages served from MongoDB. To begin:

import asyncio

from aiohttp import web
from motor.motor_asyncio import AsyncIOMotorClient


@asyncio.coroutine
def setup_db():
    db = AsyncIOMotorClient().test
    yield from db.pages.drop()
    html = '<html><body>{}</body></html>'
    yield from db.pages.insert_one({'_id': 'page-one',
                                    'body': html.format('Hello!')})

    yield from db.pages.insert_one({'_id': 'page-two',
                                    'body': html.format('Goodbye.')})

    return db

The AsyncIOMotorClient constructor does not actually connect to MongoDB. The client connects on demand, when you attempt the first operation. We create it and assign the “test” database’s handle to db.

The setup_db coroutine drops the “pages” collection (plainly, this code is for demonstration purposes), then inserts two documents. Each document’s page name is its unique id, and the “body” field is a simple HTML page. Finally, setup_db returns the database handle.

We’ll use the setup_db coroutine soon. First, we need a request handler that serves pages from the data we stored in MongoDB.

@asyncio.coroutine
def page_handler(request):
    # If the visitor gets "/pages/page-one", then page_name is "page-one".
    page_name = request.match_info.get('page_name')

    # Retrieve the long-lived database handle.
    db = request.app['db']

    # Find the page by its unique id.
    document = yield from db.pages.find_one(page_name)

    if not document:
        return web.HTTPNotFound(text='No page named {!r}'.format(page_name))

    return web.Response(body=document['body'].encode(),
                        content_type='text/html')

We start the server by running setup_db and passing the database handle to an aiohttp.web.Application:

loop = asyncio.get_event_loop()
db = loop.run_until_complete(setup_db())
app = web.Application()
app['db'] = db
# Route requests to the page_handler() coroutine.
app.router.add_get('/pages/{page_name}', page_handler)
web.run_app(app)

Note that it is a common mistake to create a new client object for every request; this comes at a dire performance cost. Create the client when your application starts and reuse that one client for the lifetime of the process. You can maintain the client by storing a database handle from the client on your application object, as shown in this example.

Visit localhost:8080/pages/page-one and the server responds “Hello!”. At localhost:8080/pages/page-two it responds “Goodbye.” At other URLs it returns a 404.

The complete code is in the Motor repository in examples/aiohttp_example.py.

See also the AIOHTTPGridFS Example.

Further Reading

The handful of classes and methods introduced here are sufficient for daily tasks. The API documentation for AsyncIOMotorClient, AsyncIOMotorDatabase, AsyncIOMotorCollection, and AsyncIOMotorCursor provides a reference to Motor’s complete feature set.

Learning to use the MongoDB driver is just the beginning, of course. For in-depth instruction in MongoDB itself, see The MongoDB Manual.