Cosmos DB Integration (Managed Identity)¶
Use this recipe to connect a Python Container App to Azure Cosmos DB (NoSQL) without account keys or connection strings.
Architecture¶
flowchart LR
C[Client] --> I[Container Apps Ingress]
I --> APP[Container App]
APP --> COSMOS[Azure Cosmos DB]
APP -.-> MI[Managed Identity]
MI -.-> ENTRA[Microsoft Entra ID]
MI -.-> COSMOS Solid arrows show runtime data flow. Dashed arrows show identity and authentication.
Prerequisites¶
- Existing Container App:
$APP_NAMEin resource group$RG - Existing Cosmos DB account and SQL database/container
- Azure CLI with Container Apps and Cosmos extensions
Step 1: Enable managed identity on the Container App¶
Get the managed identity principal ID:
export PRINCIPAL_ID=$(az containerapp show \
--name "$APP_NAME" \
--resource-group "$RG" \
--query "identity.principalId" \
--output tsv)
Step 2: Grant Cosmos DB data-plane access¶
export COSMOS_ACCOUNT_ID=$(az cosmosdb show \
--name "$COSMOS_ACCOUNT" \
--resource-group "$RG" \
--query "id" \
--output tsv)
az role assignment create \
--assignee-object-id "$PRINCIPAL_ID" \
--assignee-principal-type ServicePrincipal \
--role "Cosmos DB Built-in Data Contributor" \
--scope "$COSMOS_ACCOUNT_ID"
Step 3: Configure endpoint in Container Apps settings¶
Only store non-secret values as plain environment variables.
az containerapp update \
--name "$APP_NAME" \
--resource-group "$RG" \
--set-env-vars COSMOS_ENDPOINT="https://$COSMOS_ACCOUNT.documents.azure.com:443/" COSMOS_DATABASE="$COSMOS_DATABASE" COSMOS_CONTAINER="$COSMOS_CONTAINER"
Step 4: Python code (passwordless access)¶
Install dependencies:
Use DefaultAzureCredential in your app:
import os
from azure.identity import DefaultAzureCredential
from azure.cosmos import CosmosClient
credential = DefaultAzureCredential()
endpoint = os.environ["COSMOS_ENDPOINT"]
database_name = os.environ["COSMOS_DATABASE"]
container_name = os.environ["COSMOS_CONTAINER"]
client = CosmosClient(url=endpoint, credential=credential)
database = client.get_database_client(database_name)
container = database.get_container_client(container_name)
item = {"id": "order-1001", "type": "order", "status": "created"}
container.upsert_item(item)
stored = container.read_item(item="order-1001", partition_key="order-1001")
print(stored)
Verification steps¶
- Confirm identity assignment:
az containerapp show \
--name "$APP_NAME" \
--resource-group "$RG" \
--query "identity" \
--output json
- Confirm role assignment exists:
az role assignment list \
--assignee "$PRINCIPAL_ID" \
--scope "$COSMOS_ACCOUNT_ID" \
--output table
- Check app logs for successful read/write operations.