How to connect to MongoDB using pymongo in Python

Starting with pymongo is straightforward, but there are a few key steps to get your environment set up. First, you’ll want to ensure that you have Python installed, ideally a version that’s 3.6 or higher, as pymongo operates efficiently on these versions.

Once you have Python installed, the next step is to install pymongo itself. You can do this using pip, Python’s package manager. Just run the following command in your terminal:

pip install pymongo

This will fetch the latest version of pymongo from the Python Package Index (PyPI) and install it in your environment. If you’re using virtual environments, make sure you’ve activated it before running the command to avoid cluttering your global site-packages.

Now that you have pymongo installed, it’s a good idea to check that your MongoDB instance is up and running. You can either use a local installation of MongoDB or a cloud-based service like MongoDB Atlas. If you choose MongoDB Atlas, don’t forget to set up your cluster and whitelist your IP address so you can connect without any issues.

Now, let’s create a simple Python script to check if everything is working as expected. Start by importing pymongo in your Python file:

from pymongo import MongoClient

# Adjust the connection string as needed
client = MongoClient('mongodb://localhost:27017/')

With the initial connection established, you can verify the connection by listing the databases:

# List databases
databases = client.list_database_names()
print(databases)

This should print out an array of database names. If you see a list, congratulations, you’ve successfully connected to your MongoDB instance!

Next, you might want to create a specific database and collection to work with. You can do this in a straightforward manner:

# Create or access a database
db = client['example_database']

# Create or access a collection
collection = db['example_collection']

Now you’re ready to dive deeper into performing operations on your MongoDB instance. Understanding how to perform CRUD (Create, Read, Update, Delete) operations will solidify your knowledge of how pymongo interacts with MongoDB.

Establishing a connection to your MongoDB instance

The connection string you pass to MongoClient is actually a powerful URI that can hold much more than just a hostname and port. The full format, mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]], allows you to specify credentials, multiple hosts for a replica set, and various connection options all in one place. While it might look intimidating, it’s a standardized and efficient way to configure your database connection.

In any real-world application, you’ll be connecting to a database that requires authentication. You can embed the username and password directly into the URI. While convenient for a quick test, hardcoding credentials in your source code is a major security risk. A better approach is to load them from environment variables or a configuration management system.

# A connection string with authentication credentials
# In a real app, pull "myuser" and "mypassword" from a secure source
client = MongoClient('mongodb://myuser:mypassword@localhost:27017/')

What happens if that single MongoDB server at localhost goes down? Your application stops working. This is why production environments use a replica set-a cluster of MongoDB servers that maintain identical data sets. If the primary server fails, an election occurs, and one of the secondary servers is promoted to primary, allowing your application to continue running with minimal interruption. To connect to a replica set, you provide a list of the member hosts in the URI and specify the name of the replica set as a keyword argument.

# Connecting to a replica set for high availability
client = MongoClient(
    'mongodb://host1.example.com:27017,host2.example.com:27017/',
    replicaSet='myProductionReplicaSet'
)

When your application starts, pymongo will try to find a suitable server to connect to. If the database is unavailable, your application might hang while the driver waits. To prevent this, you should always configure a timeout. The serverSelectionTimeoutMS option tells the driver how many milliseconds to wait to find an available server before raising an error. This is crucial for building resilient applications that can handle database downtime gracefully.

You should wrap your initial connection logic in a try...except block to handle potential failures, such as a ServerSelectionTimeoutError. This is much better than letting an unhandled exception crash your entire application on startup.

from pymongo import MongoClient
from pymongo.errors import ServerSelectionTimeoutError

# A robust connection attempt with a 5-second timeout
try:
    client = MongoClient(
        'mongodb://nonexistent.host:27017/',
        serverSelectionTimeoutMS=5000
    )
    # The ismaster command is a lightweight way to force a connection
    # and verify that the server is reachable.
    client.admin.command('ismaster')
    print("MongoDB connection successful.")
except ServerSelectionTimeoutError as err:
    # Log the error and handle it gracefully
    print(f"Failed to connect to MongoDB: {err}")
    # You might exit the application or retry later

It’s important to understand that the MongoClient instance is designed to be created once and reused throughout your application. It is thread-safe and manages an internal pool of connections to the database. Creating a new MongoClient for each database operation is a common anti-pattern that is highly inefficient and will hurt your application’s performance. The connection itself is established lazily. Simply creating the client object doesn’t trigger a network I/O. The first actual operation, like a find or an insert, is what establishes the socket connections as needed.

Performing basic CRUD operations with pymongo

With a connection established and a collection object in hand, performing the fundamental database operations is refreshingly simple. MongoDB stores data in BSON format, which pymongo seamlessly maps to and from standard Python dictionaries. This means you don’t have to fiddle with object-relational mappers (ORMs) or write complex SQL strings to get work done. You just work with dictionaries.

Let’s start by creating some data. The insert_one() method adds a single document (a Python dictionary) to your collection. If your dictionary doesn’t include an _id key, MongoDB will automatically add one for you. The method returns an instance of InsertOneResult, which contains the _id of the newly inserted document.

# Assuming 'collection' is your collection object from the previous section
post = {
    "author": "Mike",
    "text": "My first blog post!",
    "tags": ["mongodb", "python", "pymongo"]
}
result = collection.insert_one(post)
print(f"One post inserted with id: {result.inserted_id}")

Inserting multiple documents at once is more efficient than inserting them one by one in a loop because it requires only a single round trip to the database. For this, you use insert_many(), passing it a list of dictionaries. It returns an InsertManyResult object containing a list of the _id values for all the new documents.

new_posts = [{
    "author": "Mike",
    "text": "Another post!",
    "tags": ["bulk", "insert"],
}, {
    "author": "Eliot",
    "text": "My first post on this platform.",
    "tags": ["mongodb", "beginner"]
}]
result = collection.insert_many(new_posts)
print(f"Multiple posts inserted with ids: {result.inserted_ids}")

Now that we have data, let’s read it. To find a single document, use find_one(). You can pass it a query filter dictionary to specify what you’re looking for. If it finds a match, it returns the document as a dictionary; otherwise, it returns None. This is perfect for fetching a user by their unique username or an article by its slug.

# Find one document written by Eliot
eliot_post = collection.find_one({"author": "Eliot"})
print(eliot_post)

To retrieve multiple documents, you use the find() method. This is where a crucial distinction lies: find() does not return a list of documents. Instead, it returns a Cursor object. A cursor is an iterable that fetches documents from the database lazily, as you iterate over it. This is incredibly memory-efficient, as it prevents you from loading, say, a million documents into your application’s memory all at once. You simply loop over the cursor.

# Find all documents written by Mike and iterate through them
for post in collection.find({"author": "Mike"}):
    print(post)

Updating documents is just as direct. The update_one() method modifies the first document that matches your query filter. You must use an update operator, like $set, to specify which fields to change. Forgetting the operator and passing a plain dictionary will replace the entire document, which is rarely what you want. The method returns an UpdateResult object, which tells you how many documents were matched and how many were actually modified.

# Find a post by Eliot and add a "comments" field to it
query = {"author": "Eliot"}
update = {"$set": {"comments": 0}}
result = collection.update_one(query, update)

print(f"Documents matched: {result.matched_count}")
print(f"Documents modified: {result.modified_count}")

If you need to modify all documents that match a query, use update_many(). For example, let’s add a new field to all of Mike’s posts.

# Add a 'status' field to all of Mike's posts
query = {"author": "Mike"}
update = {"$set": {"status": "published"}}
result = collection.update_many(query, update)

print(f"Documents matched: {result.matched_count}")
print(f"Documents modified: {result.modified_count}")

Finally, there’s deleting documents. You have delete_one() and delete_many(), which work just as you’d expect. They take a query filter and remove the first matching document or all matching documents, respectively. They return a DeleteResult object containing a deleted_count attribute.

# Delete Eliot's post
result = collection.delete_one({"author": "Eliot"})
print(f"Documents deleted: {result.deleted_count}")

# Be careful with delete_many! This will delete all of Mike's posts.
# result = collection.delete_many({"author": "Mike"})
# print(f"Documents deleted: {result.deleted_count}")

What are some of the trickier edge cases you’ve run into with CRUD operations in a production environment? Chime in with your experiences.