Azure Cosmos DB Py

1---
2name: azure-cosmos-db-py
3description: Build Azure Cosmos DB NoSQL services with Python/FastAPI following production-grade patterns. Use when implementing database client setup with dual auth (DefaultAzureCredential + emulator), service layer classes with CRUD operations, partition key strategies, parameterized queries, or TDD patterns for Cosmos. Triggers on phrases like "Cosmos DB", "NoSQL database", "document store", "add persistence", "database service layer", or "Python Cosmos SDK".
4package: azure-cosmos
5---
6 
7# Cosmos DB Service Implementation
8 
9Build production-grade Azure Cosmos DB NoSQL services following clean code, security best practices, and TDD principles.
10 
11## Installation
12 
13```bash
14pip install azure-cosmos azure-identity
15```
16 
17## Environment Variables
18 
19```bash
20COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
21COSMOS_DATABASE_NAME=<database-name>
22COSMOS_CONTAINER_ID=<container-id>
23# For emulator only (not production)
24COSMOS_KEY=<emulator-key>
25```
26 
27## Authentication
28 
29**DefaultAzureCredential (preferred)**:
30```python
31from azure.cosmos import CosmosClient
32from azure.identity import DefaultAzureCredential
33 
34client = CosmosClient(
35    url=os.environ["COSMOS_ENDPOINT"],
36    credential=DefaultAzureCredential()
37)
38```
39 
40**Emulator (local development)**:
41```python
42from azure.cosmos import CosmosClient
43 
44client = CosmosClient(
45    url="https://localhost:8081",
46    credential=os.environ["COSMOS_KEY"],
47    connection_verify=False
48)
49```
50 
51## Architecture Overview
52 
53```
54┌─────────────────────────────────────────────────────────────────┐
55│                         FastAPI Router                          │
56│  - Auth dependencies (get_current_user, get_current_user_required)
57│  - HTTP error responses (HTTPException)                         │
58└──────────────────────────────┬──────────────────────────────────┘
59                               │
60┌──────────────────────────────▼──────────────────────────────────┐
61│                        Service Layer                            │
62│  - Business logic and validation                                │
63│  - Document ↔ Model conversion                                  │
64│  - Graceful degradation when Cosmos unavailable                 │
65└──────────────────────────────┬──────────────────────────────────┘
66                               │
67┌──────────────────────────────▼──────────────────────────────────┐
68│                     Cosmos DB Client Module                     │
69│  - Singleton container initialization                           │
70│  - Dual auth: DefaultAzureCredential (Azure) / Key (emulator)   │
71│  - Async wrapper via run_in_threadpool                          │
72└─────────────────────────────────────────────────────────────────┘
73```
74 
75## Quick Start
76 
77### 1. Client Module Setup
78 
79Create a singleton Cosmos client with dual authentication:
80 
81```python
82# db/cosmos.py
83from azure.cosmos import CosmosClient
84from azure.identity import DefaultAzureCredential
85from starlette.concurrency import run_in_threadpool
86 
87_cosmos_container = None
88 
89def _is_emulator_endpoint(endpoint: str) -> bool:
90    return "localhost" in endpoint or "127.0.0.1" in endpoint
91 
92async def get_container():
93    global _cosmos_container
94    if _cosmos_container is None:
95        if _is_emulator_endpoint(settings.cosmos_endpoint):
96            client = CosmosClient(
97                url=settings.cosmos_endpoint,
98                credential=settings.cosmos_key,
99                connection_verify=False
100            )
101        else:
102            client = CosmosClient(
103                url=settings.cosmos_endpoint,
104                credential=DefaultAzureCredential()
105            )
106        db = client.get_database_client(settings.cosmos_database_name)
107        _cosmos_container = db.get_container_client(settings.cosmos_container_id)
108    return _cosmos_container
109```
110 
111**Full implementation**: See [references/client-setup.md](references/client-setup.md)
112 
113### 2. Pydantic Model Hierarchy
114 
115Use five-tier model pattern for clean separation:
116 
117```python
118class ProjectBase(BaseModel):           # Shared fields
119    name: str = Field(..., min_length=1, max_length=200)
120 
121class ProjectCreate(ProjectBase):       # Creation request
122    workspace_id: str = Field(..., alias="workspaceId")
123 
124class ProjectUpdate(BaseModel):         # Partial updates (all optional)
125    name: Optional[str] = Field(None, min_length=1)
126 
127class Project(ProjectBase):             # API response
128    id: str
129    created_at: datetime = Field(..., alias="createdAt")
130 
131class ProjectInDB(Project):             # Internal with docType
132    doc_type: str = "project"
133```
134 
135### 3. Service Layer Pattern
136 
137```python
138class ProjectService:
139    def _use_cosmos(self) -> bool:
140        return get_container() is not None
141    
142    async def get_by_id(self, project_id: str, workspace_id: str) -> Project | None:
143        if not self._use_cosmos():
144            return None
145        doc = await get_document(project_id, partition_key=workspace_id)
146        if doc is None:
147            return None
148        return self._doc_to_model(doc)
149```
150 
151**Full patterns**: See [references/service-layer.md](references/service-layer.md)
152 
153## Core Principles
154 
155### Security Requirements
156 
1571. **RBAC Authentication**: Use `DefaultAzureCredential` in Azure — never store keys in code
1582. **Emulator-Only Keys**: Hardcode the well-known emulator key only for local development
1593. **Parameterized Queries**: Always use `@parameter` syntax — never string concatenation
1604. **Partition Key Validation**: Validate partition key access matches user authorization
161 
162### Clean Code Conventions
163 
1641. **Single Responsibility**: Client module handles connection; services handle business logic
1652. **Graceful Degradation**: Services return `None`/`[]` when Cosmos unavailable
1663. **Consistent Naming**: `_doc_to_model()`, `_model_to_doc()`, `_use_cosmos()`
1674. **Type Hints**: Full typing on all public methods
1685. **CamelCase Aliases**: Use `Field(alias="camelCase")` for JSON serialization
169 
170### TDD Requirements
171 
172Write tests BEFORE implementation using these patterns:
173 
174```python
175@pytest.fixture
176def mock_cosmos_container(mocker):
177    container = mocker.MagicMock()
178    mocker.patch("app.db.cosmos.get_container", return_value=container)
179    return container
180 
181@pytest.mark.asyncio
182async def test_get_project_by_id_returns_project(mock_cosmos_container):
183    # Arrange
184    mock_cosmos_container.read_item.return_value = {"id": "123", "name": "Test"}
185    
186    # Act
187    result = await project_service.get_by_id("123", "workspace-1")
188    
189    # Assert
190    assert result.id == "123"
191    assert result.name == "Test"
192```
193 
194**Full testing guide**: See [references/testing.md](references/testing.md)
195 
196## Reference Files
197 
198| File | When to Read |
199|------|--------------|
200| [references/client-setup.md](references/client-setup.md) | Setting up Cosmos client with dual auth, SSL config, singleton pattern |
201| [references/service-layer.md](references/service-layer.md) | Implementing full service class with CRUD, conversions, graceful degradation |
202| [references/testing.md](references/testing.md) | Writing pytest tests, mocking Cosmos, integration test setup |
203| [references/partitioning.md](references/partitioning.md) | Choosing partition keys, cross-partition queries, move operations |
204| [references/error-handling.md](references/error-handling.md) | Handling CosmosResourceNotFoundError, logging, HTTP error mapping |
205 
206## Template Files
207 
208| File | Purpose |
209|------|---------|
210| [assets/cosmos_client_template.py](assets/cosmos_client_template.py) | Ready-to-use client module |
211| [assets/service_template.py](assets/service_template.py) | Service class skeleton |
212| [assets/conftest_template.py](assets/conftest_template.py) | pytest fixtures for Cosmos mocking |
213 
214## Quality Attributes (NFRs)
215 
216### Reliability
217- Graceful degradation when Cosmos unavailable
218- Retry logic with exponential backoff for transient failures
219- Connection pooling via singleton pattern
220 
221### Security
222- Zero secrets in code (RBAC via DefaultAzureCredential)
223- Parameterized queries prevent injection
224- Partition key isolation enforces data boundaries
225 
226### Maintainability
227- Five-tier model pattern enables schema evolution
228- Service layer decouples business logic from storage
229- Consistent patterns across all entity services
230 
231### Testability
232- Dependency injection via `get_container()`
233- Easy mocking with module-level globals
234- Clear separation enables unit testing without Cosmos
235 
236### Performance
237- Partition key queries avoid cross-partition scans
238- Async wrapping prevents blocking FastAPI event loop
239- Minimal document conversion overhead
240