Build production-grade Azure Cosmos DB NoSQL services following clean code, security best practices, and TDD principles.
\npip install azure-cosmos azure-identity\nCOSMOS_ENDPOINT=https://<account>.documents.azure.com:443/ # Required for all auth methods\nCOSMOS_DATABASE_NAME=<database-name> # Required for all auth methods\nCOSMOS_CONTAINER_ID=<container-id> # Required for all auth methods\n# For emulator only (not production)\nCOSMOS_KEY=<emulator-key> # Only required for key-based auth or emulator\nAZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production\n\n\n\uD83D\uDD11 Two rules apply to every code sample below:
\n\n
\n- Prefer
\nDefaultAzureCredential. 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.\n\n
\n- Local dev:
\nDefaultAzureCredentialworks as-is.- Production: set
\nAZURE_TOKEN_CREDENTIALS=prod(orAZURE_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:\n
\n\n
\n- Sync:
\nwith <Client>(...) as client:- Async:
\nasync with <Client>(...) as client:andasync with DefaultAzureCredential() as credential:(fromazure.identity.aio)Snippets may abbreviate this setup, but production code should always follow both rules.
\n
DefaultAzureCredential (preferred):
\nimport os\nfrom azure.cosmos import CosmosClient\nfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredential\n\n# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>\ncredential = DefaultAzureCredential(require_envvar=True)\n# Or use a specific credential directly in production:\n# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes\n# credential = ManagedIdentityCredential()\n\nwith CosmosClient(\n url=os.environ[\"COSMOS_ENDPOINT\"],\n credential=credential\n) as client:\n # Use client here (see following sections for operations)\n ...\nEmulator (local development):
\nfrom azure.cosmos import CosmosClient\n\nwith CosmosClient(\n url=\"https://localhost:8081\",\n credential=os.environ[\"COSMOS_KEY\"],\n connection_verify=False\n) as client:\n # Use client here (see following sections for operations)\n ...\n┌─────────────────────────────────────────────────────────────────┐\n│ FastAPI Router │\n│ - Auth dependencies (get_current_user, get_current_user_required)\n│ - HTTP error responses (HTTPException) │\n└──────────────────────────────┬──────────────────────────────────┘\n │\n┌──────────────────────────────▼──────────────────────────────────┐\n│ Service Layer │\n│ - Business logic and validation │\n│ - Document ↔ Model conversion │\n│ - Graceful degradation when Cosmos unavailable │\n└──────────────────────────────┬──────────────────────────────────┘\n │\n┌──────────────────────────────▼──────────────────────────────────┐\n│ Cosmos DB Client Module │\n│ - Singleton container initialization │\n│ - Dual auth: DefaultAzureCredential (Azure) / Key (emulator) │\n│ - Async wrapper via run_in_threadpool │\n└─────────────────────────────────────────────────────────────────┘\nCreate a singleton Cosmos client with dual authentication:
\n# db/cosmos.py\nfrom azure.cosmos import CosmosClient\nfrom azure.identity import DefaultAzureCredential\nfrom starlette.concurrency import run_in_threadpool\n\n_cosmos_container = None\n\ndef _is_emulator_endpoint(endpoint: str) -> bool:\n return \"localhost\" in endpoint or \"127.0.0.1\" in endpoint\n\nasync def get_container():\n global _cosmos_container\n if _cosmos_container is None:\n # Singleton: client lives for the FastAPI app lifetime; close in a lifespan shutdown handler.\n if _is_emulator_endpoint(settings.cosmos_endpoint):\n client = CosmosClient(\n url=settings.cosmos_endpoint,\n credential=settings.cosmos_key,\n connection_verify=False\n )\n else:\n client = CosmosClient(\n url=settings.cosmos_endpoint,\n credential=DefaultAzureCredential()\n )\n db = client.get_database_client(settings.cosmos_database_name)\n _cosmos_container = db.get_container_client(settings.cosmos_container_id)\n return _cosmos_container\nFull implementation: See references/client-setup.md
\nUse five-tier model pattern for clean separation:
\nclass ProjectBase(BaseModel): # Shared fields\n name: str = Field(..., min_length=1, max_length=200)\n\nclass ProjectCreate(ProjectBase): # Creation request\n workspace_id: str = Field(..., alias=\"workspaceId\")\n\nclass ProjectUpdate(BaseModel): # Partial updates (all optional)\n name: Optional[str] = Field(None, min_length=1)\n\nclass Project(ProjectBase): # API response\n id: str\n created_at: datetime = Field(..., alias=\"createdAt\")\n\nclass ProjectInDB(Project): # Internal with docType\n doc_type: str = \"project\"\nclass ProjectService:\n def _use_cosmos(self) -> bool:\n return get_container() is not None\n \n async def get_by_id(self, project_id: str, workspace_id: str) -> Project | None:\n if not self._use_cosmos():\n return None\n doc = await get_document(project_id, partition_key=workspace_id)\n if doc is None:\n return None\n return self._doc_to_model(doc)\nFull patterns: See references/service-layer.md
\nDefaultAzureCredential in Azure — never store keys in code@parameter syntax — never string concatenationNone/[] when Cosmos unavailable_doc_to_model(), _model_to_doc(), _use_cosmos()Field(alias=\"camelCase\") for JSON serializationWrite tests BEFORE implementation using these patterns:
\n@pytest.fixture\ndef mock_cosmos_container(mocker):\n container = mocker.MagicMock()\n mocker.patch(\"app.db.cosmos.get_container\", return_value=container)\n return container\n\n@pytest.mark.asyncio\nasync def test_get_project_by_id_returns_project(mock_cosmos_container):\n # Arrange\n mock_cosmos_container.read_item.return_value = {\"id\": \"123\", \"name\": \"Test\"}\n \n # Act\n result = await project_service.get_by_id(\"123\", \"workspace-1\")\n \n # Assert\n assert result.id == \"123\"\n assert result.name == \"Test\"\nFull testing guide: See references/testing.md
\nazure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.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.| File | \nWhen to Read | \n
|---|---|
| references/client-setup.md | \nSetting up Cosmos client with dual auth, SSL config, singleton pattern | \n
| references/service-layer.md | \nImplementing full service class with CRUD, conversions, graceful degradation | \n
| references/testing.md | \nWriting pytest tests, mocking Cosmos, integration test setup | \n
| references/partitioning.md | \nChoosing partition keys, cross-partition queries, move operations | \n
| references/error-handling.md | \nHandling CosmosResourceNotFoundError, logging, HTTP error mapping | \n
| File | \nPurpose | \n
|---|---|
| assets/cosmos_client_template.py | \nReady-to-use client module | \n
| assets/service_template.py | \nService class skeleton | \n
| assets/conftest_template.py | \npytest fixtures for Cosmos mocking | \n
get_container()