azure-cosmos-db-py

by microsoft

Build Azure Cosmos DB NoSQL services with Python/FastAPI following production-grade patterns. Use when implementing database client setup with dual auth… npx skills add https://github.com/microsoft/agent-skills --skill azure-cosmos-db-py Download ZIPGitHub2.7k

Cosmos DB Service Implementation

Build production-grade Azure Cosmos DB NoSQL services following clean code, security best practices, and TDD principles.

Environment Variables

COSMOS_ENDPOINT=https:// .documents.azure.com:443/ # Required for all auth methods
COSMOS_DATABASE_NAME= # Required for all auth methods
COSMOS_CONTAINER_ID= # Required for all auth methods
# For emulator only (not production)
COSMOS_KEY= # Only required for key-based auth or emulator
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

• Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.

• Local dev: DefaultAzureCredential works as-is.

• Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.

• Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:

• Sync: with <Client>(...) as client:

• Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)

Snippets may abbreviate this setup, but production code should always follow both rules.

DefaultAzureCredential (preferred):

import os
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS= 
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

with CosmosClient(
 url=os.environ["COSMOS_ENDPOINT"],
 credential=credential
) as client:
 # Use client here (see following sections for operations)
 ...

Emulator (local development):

from azure.cosmos import CosmosClient

with CosmosClient(
 url="https://localhost:8081",
 credential=os.environ["COSMOS_KEY"],
 connection_verify=False
) as client:
 # Use client here (see following sections for operations)
 ...

Architecture Overview

┌─────────────────────────────────────────────────────────────────┐
│ FastAPI Router │
│ - Auth dependencies (get_current_user, get_current_user_required)
│ - HTTP error responses (HTTPException) │
└──────────────────────────────┬──────────────────────────────────┘
 │
┌──────────────────────────────▼──────────────────────────────────┐
│ Service Layer │
│ - Business logic and validation │
│ - Document ↔ Model conversion │
│ - Graceful degradation when Cosmos unavailable │
└──────────────────────────────┬──────────────────────────────────┘
 │
┌──────────────────────────────▼──────────────────────────────────┐
│ Cosmos DB Client Module │
│ - Singleton container initialization │
│ - Dual auth: DefaultAzureCredential (Azure) / Key (emulator) │
│ - Async wrapper via run_in_threadpool │
└─────────────────────────────────────────────────────────────────┘

Core Principles

Security Requirements

• RBAC Authentication: Use DefaultAzureCredential in Azure — never store keys in code

• Emulator-Only Keys: Hardcode the well-known emulator key only for local development

• Parameterized Queries: Always use @parameter syntax — never string concatenation

• Partition Key Validation: Validate partition key access matches user authorization

Clean Code Conventions

• Single Responsibility: Client module handles connection; services handle business logic

• Graceful Degradation: Services return None/[] when Cosmos unavailable

• Consistent Naming: _doc_to_model(), _model_to_doc(), _use_cosmos()

• Type Hints: Full typing on all public methods

• CamelCase Aliases: Use Field(alias="camelCase") for JSON serialization

TDD Requirements

Write tests BEFORE implementation using these patterns:

@pytest.fixture
def mock_cosmos_container(mocker):
 container = mocker.MagicMock()
 mocker.patch("app.db.cosmos.get_container", return_value=container)
 return container

@pytest.mark.asyncio
async def test_get_project_by_id_returns_project(mock_cosmos_container):
 # Arrange
 mock_cosmos_container.read_item.return_value = {"id": "123", "name": "Test"}
 
 # Act
 result = await project_service.get_by_id("123", "workspace-1")
 
 # Assert
 assert result.id == "123"
 assert result.name == "Test"

Full testing guide: See references/testing.md

Best Practices

• This skill uses async throughout (azure.cosmos.aio); do not mix with the sync azure.cosmos client. Keep the whole FastAPI request path async — don't pair sync Cosmos calls with async handlers.

• Always use context managers for clients and async credentials. Wrap the client in async with CosmosClient(...) as client: (or manage its lifetime via FastAPI lifespan and close it explicitly). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.

Reference Files

File When to Read references/client-setup.md Setting up Cosmos client with dual auth, SSL config, singleton pattern references/service-layer.md Implementing full service class with CRUD, conversions, graceful degradation references/testing.md Writing pytest tests, mocking Cosmos, integration test setup references/partitioning.md Choosing partition keys, cross-partition queries, move operations references/error-handling.md Handling CosmosResourceNotFoundError, logging, HTTP error mapping

Template Files

File Purpose assets/cosmos_client_template.py Ready-to-use client module assets/service_template.py Service class skeleton assets/conftest_template.py pytest fixtures for Cosmos mocking

Quality Attributes (NFRs)

Reliability

• Graceful degradation when Cosmos unavailable

• Retry logic with exponential backoff for transient failures

• Connection pooling via singleton pattern

Security

• Zero secrets in code (RBAC via DefaultAzureCredential)

• Parameterized queries prevent injection

• Partition key isolation enforces data boundaries

Maintainability

• Five-tier model pattern enables schema evolution

• Service layer decouples business logic from storage

• Consistent patterns across all entity services

Testability

• Dependency injection via get_container()

• Easy mocking with module-level globals

• Clear separation enables unit testing without Cosmos

Performance

• Partition key queries avoid cross-partition scans

• Async wrapping prevents blocking FastAPI event loop

• Minimal document conversion overhead