Implement Production-Ready Azure Cosmos DB Services
Skill for building production-grade Azure Cosmos DB services in Python with clean architecture and TDD.
Why it matters
Build robust and secure Azure Cosmos DB NoSQL services using clean code and Test-Driven Development (TDD) principles. This asset provides a structured approach to implementing database interactions, ensuring reliability and maintainability.
Outcomes
What it gets done
Implement a singleton Cosmos DB client with dual authentication for Azure and local emulator.
Develop a service layer with business logic, validation, and graceful degradation.
Define Pydantic models for data representation and API interactions.
Write comprehensive unit and integration tests following TDD practices.
Install
Add it to your toolbox
Run in your project directory:
curl -fsSL https://spark.entire.vc/get/ag-azure-cosmos-db-py | bash Overview
Cosmos DB Service Implementation
Skill for building production-grade Azure Cosmos DB services in Python, covering RBAC authentication with emulator fallback, a five-tier Pydantic model pattern, a graceful-degradation service layer over a singleton Cosmos client, and TDD with mocked Cosmos containers. Use when building a production, RBAC-secured, test-driven Cosmos DB service in Python, typically behind FastAPI - not for ad hoc scripts.
What it does
This skill builds production-grade Azure Cosmos DB NoSQL services in Python (pip install azure-cosmos azure-identity), structured around clean code, security best practices, and test-driven development. It authenticates via DefaultAzureCredential in Azure (preferred, RBAC-based, no stored keys) or a well-known emulator key for local development against https://localhost:8081, and its reference architecture layers a FastAPI router (auth dependencies, HTTP error responses) over a service layer (business logic, document-to-model conversion, graceful degradation when Cosmos is unavailable) over a Cosmos DB client module (singleton container initialization, dual auth switching between Azure and emulator, and an async wrapper via run_in_threadpool since the Cosmos SDK itself is synchronous). The quick-start client module detects an emulator endpoint by checking for localhost/127.0.0.1 and picks the right credential path, caching the container client as a module-level singleton so it's only initialized once. Data modeling follows a five-tier Pydantic pattern: a Base model for shared fields, a Create model for creation requests, an Update model with all fields optional for partial updates, a response model for the API surface, and an InDB model adding internal fields like a doc_type discriminator - with camelCase JSON aliases (Field(alias="camelCase")) bridging Python's snake_case and typical JSON API conventions. The service layer pattern checks whether Cosmos is actually available before each operation and returns None/empty results gracefully rather than raising when it isn't. Security requirements are explicit: RBAC via DefaultAzureCredential with no keys in code (except the well-known emulator key, local-only), always parameterized @parameter queries (never string concatenation, to prevent injection), and partition key access validated against user authorization. Clean code conventions specify single-responsibility separation between the client module and services, consistent naming (_doc_to_model(), _model_to_doc(), _use_cosmos()), full type hints, and camelCase aliasing. TDD requirements call for writing pytest tests before implementation, using mocker.patch to mock the container-getter function and pytest.mark.asyncio for async service methods. Five companion reference files (client setup, service layer, testing, partitioning, error handling) and three ready-to-use template files (client module, service skeleton, pytest fixtures) supply the full implementation detail beyond this overview, and the skill closes with explicit quality attributes across reliability (graceful degradation, retry with backoff, singleton connection pooling), security (zero secrets in code, parameterized queries, partition key isolation), maintainability (five-tier models, decoupled service layer), testability (dependency injection via get_container(), easy mocking), and performance (partition-key queries avoiding cross-partition scans, async wrapping to avoid blocking the FastAPI event loop).
When to use - and when NOT to
Use this skill when building a production Cosmos DB-backed service in Python (typically behind FastAPI) that needs RBAC authentication, clean separation between HTTP/service/data layers, and test coverage written before implementation. It is not a guide for ad hoc scripts or one-off queries against Cosmos - the five-tier model and service-layer patterns are overhead that only pays off for a maintained, tested production service.
Inputs and outputs
Input is entity definitions (via the five-tier Pydantic model pattern) and Cosmos connection settings (endpoint, database, container, and emulator key if local). Output is a working, tested service module with a singleton Cosmos client, CRUD methods with graceful degradation, and companion reference/template files for setup, testing, partitioning, and error handling.
Integrations
Built on the azure-cosmos and azure-identity Python SDKs (CosmosClient, DefaultAzureCredential), designed for a FastAPI application layer, with pytest/pytest-asyncio/pytest-mock for TDD.
pip install azure-cosmos azure-identity
Who it's for
Python/FastAPI developers building production Cosmos DB services who need RBAC-secured, cleanly layered, test-driven data access rather than ad hoc scripts or direct SDK calls scattered through route handlers.
FAQ
Common questions
Discussion
Questions & comments ยท 0
Sign In Sign in to leave a comment.