Motor is an asynchronous Python driver for MongoDB. It integrates with asyncio to perform nonblocking database operations, making it suited to web servers, real-time processors, and applications handling many concurrent I/O tasks. section 3.6 section 3.9
Choose Motor when application behavior and workload justify asynchronous database access:
- Prefer Motor for I/O-bound applications that perform simultaneous database operations, network requests, or long-running waits. section 3.6.3
- Prefer PyMongo for synchronous scripts, sequential tasks, manageable workloads, or applications where simpler control flow matters more than concurrency. section 3.6 section 3.6.3
- Account for async/await patterns and event-loop behavior, because asynchronous execution improves responsiveness but adds complexity. section 3.6.3
Query a route asynchronously
Install Motor with pip, place the Atlas connection string in a Python script, create an AsyncIOMotorClient, select sample_training.routes, and await find_one({"src_airport": "JFK", "airline.id": 3201}). Print the matching route or a not-found message, handle errors, and close the connection efficiently. Run the script with python3 mongodb-motor.py. section 3.6.2
Listing 3.10 Motor Python driver example
import asyncio #1
from motor.motor_asyncio import AsyncIOMotorClient #1
from pymongo.server_api import ServerApi #1
from pprint import pprint #1
# Replace <connection_string> with your MongoDB Atlas connection string.
uri = "<connection_string>" #2
async def find_route(): #3
# Create an AsyncIOMotorClient instance specifying the Stable API version
client = AsyncIOMotorClient(uri, server_api=ServerApi('1'))
➥#4
try:
# Specify the database and collection
database = client['sample_training'] #5
routes_collection = database['routes'] #5
# Define the query
query = {"src_airport": "JFK", "airline.id": 3201}
➥#6
# Execute the query to find a single document matching the query
route = await routes_collection.find_one(query) #7
# Check if a route is found and print the result
if route:
print("Found a route:") #8
pprint(route) #8
else:
print("No route found from JFK.") #9
except Exception as e:
# Handle any errors that occur during the query
print(f"An error occurred: {e}") #10
finally:
# Close the connection to MongoDB
client.close() #11
# Run the asynchronous find_route function
asyncio.run(find_route()) #12
Shows the complete Motor driver pattern, including asyncio, AsyncIOMotorClient, Stable API version 1, database and collection selection, an awaited query, and result handling. section 3.6.2
python3 -m pip install motor
Provides the pip command required to install the Motor module. section 3.6.2
