Edit

Share via

Build a Python console app with Azure Cosmos DB for MongoDB vCore

  • Applies to: ✅ MongoDB (vCore)

This guide walks you through building a Python console application to connect to an Azure Cosmos DB for MongoDB vCore cluster. You configure your development environment, authenticate using the azure.identity package from the Azure SDK for Python, and perform operations such as creating, querying, and updating documents.

Prerequisites

  • An existing Azure Cosmos DB for MongoDB (vCore) cluster.

  • Microsoft Entra authentication configured for the cluster with your identity granted dbOwner role.

  • Latest version of Python.

Configure your console application

Next, create a new console application project and import the necessary libraries to authenticate to your cluster.

  1. Create a new directory for your project and set up a virtual environment.

    mkdir cosmos-mongodb-app
cd cosmos-mongodb-app
python -m venv .venv

  2. Activate the virtual environment.

    # On Windows
.venv\Scripts\activate

# On macOS/Linux
source .venv/bin/activate

  3. Create a new Python file for your application.

    touch app.py

  4. Install the azure.identity library for Azure authentication.

    pip install azure.identity

  5. Install the pymongo driver for Python.

    pip install pymongo

Connect to the cluster

Now, use the Azure.Identity library to get a TokenCredential to use to connect to your cluster. The official MongoDB driver has a special interface that must be implemented to obtain tokens from Microsoft Entra for use when connecting to the cluster.

  1. Import the necessary modules at the top of your Python file.

    from azure.identity import DefaultAzureCredential
from pymongo import MongoClient
from pymongo.auth_oidc import OIDCCallback, OIDCCallbackContext, OIDCCallbackResult

  2. Create a custom class that implements the MongoDB OpenID Connect (OIDC) callback interface.

    class AzureIdentityTokenCallback(OIDCCallback):
    def __init__(self, credential):
        self.credential = credential

    def fetch(self, context: OIDCCallbackContext) -> OIDCCallbackResult:
        token = self.credential.get_token(
            "https://ossrdbms-aad.database.windows.net/.default").token
        return OIDCCallbackResult(access_token=token)

  3. Set your cluster name variable.

    clusterName = "<azure-cosmos-db-mongodb-vcore-cluster-name>"

  4. Create an instance of DefaultAzureCredential and set up the authentication properties.

    credential = DefaultAzureCredential()
authProperties = {"OIDC_CALLBACK": AzureIdentityTokenCallback(credential)}

  5. Create a MongoDB client configured with Microsoft Entra authentication.

    client = MongoClient(
    f"mongodb+srv://{clusterName}.global.mongocluster.cosmos.azure.com/",
    connectTimeoutMS=120000,
    tls=True,
    retryWrites=True,
    authMechanism="MONGODB-OIDC",
    authMechanismProperties=authProperties
)

print("Client created")

Perform common operations

Finally, use the official library to perform common tasks with databases, collections, and documents. Here, you use the same classes and methods you would use to interact with MongoDB or DocumentDB to manage your collections and items.

  1. Get a reference to your database.

    database = client.get_database("<database-name>")

print("Database pointer created")

  2. Get a reference to your collection.

    collection = database.get_collection("<container-name>")

print("Collection pointer created")

  3. Create a document and upsert it into the collection with collection.update_one.

    new_document = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
    "category": "gear-surf-surfboards",
    "name": "Yamba Surfboard",
    "quantity": 12,
    "price": 850.00,
    "clearance": False,
}

filter = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
}
payload = {
    "$set": new_document
}
result = collection.update_one(filter, payload, upsert=True)

  4. Use collection.find_one to retrieve a specific document from the collection.

    filter = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
    "category": "gear-surf-surfboards"
}
existing_document = collection.find_one(filter)
print(f"Read document _id:\t{existing_document['_id']}")

  5. Query for multiple documents with collection.find that matches a filter.

    filter = {
    "category": "gear-surf-surfboards"
}
matched_documents = collection.find(filter)

for document in matched_documents:
    print(f"Found document:\t{document}")

Related content