Azure Cosmos DB Py

MicrosoftOfficialCreated by Microsoft

Skill
Tool integration
Software development
MIT
Low risk
Has scripts

From the author

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".

— the description by Microsoft

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

This skill provides patterns and templates for integrating Azure Cosmos DB into Python and FastAPI applications. It focuses on secure authentication using DefaultAzureCredential, singleton client management, and service layer patterns for CRUD operations. The guidance includes strategies for partition key management, parameterized queries, and TDD-based testing using pytest.

What it does

Provides a singleton pattern for Cosmos DB client initialization.
Implements dual authentication for local emulator and Azure production environments.
Wraps synchronous SDK calls for asynchronous execution in FastAPI.
Defines a service layer pattern for document-to-model conversion.
Offers templates for pytest fixtures to facilitate unit testing.
Outlines error handling patterns for common Cosmos DB exceptions.

When to use it

When building a FastAPI backend that requires NoSQL persistence with Azure Cosmos DB.
When you need to implement secure, RBAC-based authentication for database access.
When you want to structure your database interaction layer using service classes.
When you need to write unit tests for code that interacts with the Cosmos DB SDK.

Advantages

Promotes secure coding by using identity-based authentication instead of hardcoded keys.
Ensures consistent connection management through singleton patterns.
Facilitates testability by providing patterns for mocking database containers.
Encourages clean code through separation of concerns between business logic and storage.

Usage examples

“How do I set up a singleton Cosmos DB client in my FastAPI app?”: The skill provides a module-level initialization pattern that uses DefaultAzureCredential for production and key-based auth for the local emulator.
“Show me how to test a service layer method that reads from Cosmos DB.”: The skill provides a pytest fixture template that uses mocking to simulate the Cosmos container, allowing you to test service logic without a live database.

Install

Download the skill as a ZIP:
https://autogineer.com/agent-skills/microsoft/azure-cosmos-db-py/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip azure-cosmos-db-py.zip -d .agents/skills

Documentation by Microsoft

Cosmos DB Service Implementation

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

Installation

pip install azure-cosmos azure-identity

Environment Variables

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Required for all auth methods
COSMOS_DATABASE_NAME=<database-name>  # Required for all auth methods
COSMOS_CONTAINER_ID=<container-id>  # Required for all auth methods
# For emulator only (not production)
COSMOS_KEY=<emulator-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=<specific_credential>
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                          │
└─────────────────────────────────────────────────────────────────┘

Quick Start

1. Client Module Setup

Create a singleton Cosmos client with dual authentication:

# db/cosmos.py
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential
from starlette.concurrency import run_in_threadpool

_cosmos_container = None

def _is_emulator_endpoint(endpoint: str) -> bool:
    return "localhost" in endpoint or "127.0.0.1" in endpoint

async def get_container():
    global _cosmos_container
    if _cosmos_container is None:
        # Singleton: client lives for the FastAPI app lifetime; close in a lifespan shutdown handler.
        if _is_emulator_endpoint(settings.cosmos_endpoint):
            client = CosmosClient(
                url=settings.cosmos_endpoint,
                credential=settings.cosmos_key,
                connection_verify=False
            )
        else:
            client = CosmosClient(
                url=settings.cosmos_endpoint,
                credential=DefaultAzureCredential()
            )
        db = client.get_database_client(settings.cosmos_database_name)
        _cosmos_container = db.get_container_client(settings.cosmos_container_id)
    return _cosmos_container

Full implementation: See references/client-setup.md

2. Pydantic Model Hierarchy

Use five-tier model pattern for clean separation:

class ProjectBase(BaseModel):           # Shared fields
    name: str = Field(..., min_length=1, max_length=200)

class ProjectCreate(ProjectBase):       # Creation request
    workspace_id: str = Field(..., alias="workspaceId")

class ProjectUpdate(BaseModel):         # Partial updates (all optional)
    name: Optional[str] = Field(None, min_length=1)

class Project(ProjectBase):             # API response
    id: str
    created_at: datetime = Field(..., alias="createdAt")

class ProjectInDB(Project):             # Internal with docType
    doc_type: str = "project"

3. Service Layer Pattern

class ProjectService:
    def _use_cosmos(self) -> bool:
        return get_container() is not None
    
    async def get_by_id(self, project_id: str, workspace_id: str) -> Project | None:
        if not self._use_cosmos():
            return None
        doc = await get_document(project_id, partition_key=workspace_id)
        if doc is None:
            return None
        return self._doc_to_model(doc)

Full patterns: See references/service-layer.md

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

Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.
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

Files

ATTRIBUTION.mdAdded by AutoGineer965 B
LICENSELicence file added by AutoGineer1.1 KiB
SKILL.md11.6 KiB
assets/conftest_template.pyScript9.7 KiB
assets/cosmos_client_template.pyScript6.2 KiB
assets/service_template.pyScript9.2 KiB
references/client-setup.md6.1 KiB
references/error-handling.md10.3 KiB
references/partitioning.md7.2 KiB
references/service-layer.md8.2 KiB
references/testing.md14.3 KiB

references/testing.md

Testing Cosmos DB Services

Test-Driven Development Workflow
Test File Structure
Fixtures and Mocking
Unit Test Patterns
Integration Test Patterns
Test Data Factories

Test-Driven Development Workflow

Follow Red-Green-Refactor:

Red: Write a failing test for the behavior you want
Green: Write minimal code to make the test pass
Refactor: Clean up while keeping tests green

Example TDD Cycle

# Step 1: RED - Write failing test
@pytest.mark.asyncio
async def test_create_project_generates_unique_id(mock_cosmos, project_create_data):
    result = await project_service.create(project_create_data, author_id="user-1")
    
    assert result.id is not None
    assert len(result.id) == 36  # UUID format

# Step 2: GREEN - Implement minimal code
async def create(self, data: ProjectCreate, author_id: str) -> Project:
    project_id = str(uuid.uuid4())
    # ... rest of implementation

# Step 3: REFACTOR - Clean up if needed

Test File Structure

Organize tests to mirror source structure:

tests/
├── conftest.py              # Shared fixtures
├── unit/
│   ├── services/
│   │   ├── test_project_service.py
│   │   ├── test_workspace_service.py
│   │   └── test_flow_service.py
│   └── db/
│       └── test_cosmos.py
└── integration/
    └── test_cosmos_integration.py

Test File Template

"""Tests for ProjectService."""
import pytest
from datetime import datetime, timezone
from unittest.mock import MagicMock

from app.services.project_service import project_service
from app.models.project import ProjectCreate, ProjectUpdate


class TestProjectServiceCreate:
    """Tests for ProjectService.create()"""
    
    @pytest.mark.asyncio
    async def test_create_returns_project_with_generated_id(self, mock_cosmos):
        ...
    
    @pytest.mark.asyncio
    async def test_create_sets_timestamps(self, mock_cosmos):
        ...
    
    @pytest.mark.asyncio
    async def test_create_generates_unique_slug(self, mock_cosmos):
        ...


class TestProjectServiceGetById:
    """Tests for ProjectService.get_by_id()"""
    
    @pytest.mark.asyncio
    async def test_get_by_id_returns_project_when_found(self, mock_cosmos):
        ...
    
    @pytest.mark.asyncio
    async def test_get_by_id_returns_none_when_not_found(self, mock_cosmos):
        ...
    
    @pytest.mark.asyncio
    async def test_get_by_id_returns_none_when_cosmos_unavailable(self):
        ...

Fixtures and Mocking

Core Fixtures (conftest.py)

import pytest
from unittest.mock import MagicMock, AsyncMock, patch
from datetime import datetime, timezone


@pytest.fixture
def mock_cosmos_container():
    """Mock Cosmos container with common operations."""
    container = MagicMock()
    
    # Default behaviors
    container.read_item.return_value = None
    container.upsert_item.return_value = {}
    container.delete_item.return_value = None
    container.query_items.return_value = iter([])
    
    return container


@pytest.fixture
def mock_cosmos(mock_cosmos_container, mocker):
    """Patch get_container to return mock."""
    mocker.patch(
        "app.db.cosmos.get_container",
        return_value=mock_cosmos_container
    )
    return mock_cosmos_container


@pytest.fixture
def mock_cosmos_unavailable(mocker):
    """Simulate Cosmos being unavailable."""
    mocker.patch(
        "app.db.cosmos.get_container",
        return_value=None
    )


@pytest.fixture
def sample_project_doc():
    """Sample project document as stored in Cosmos."""
    return {
        "id": "proj-123",
        "name": "Test Project",
        "description": "A test project",
        "slug": "test-project",
        "workspaceId": "ws-456",
        "authorId": "user-789",
        "visibility": "public",
        "tags": ["test", "sample"],
        "createdAt": "2024-01-15T10:30:00+00:00",
        "updatedAt": None,
        "docType": "project",
    }


@pytest.fixture
def project_create_data():
    """Sample ProjectCreate for testing."""
    return ProjectCreate(
        name="New Project",
        description="Description",
        workspace_id="ws-456",
        visibility="public",
        tags=["new"],
    )

Mocking Async Operations

@pytest.fixture
def mock_cosmos_async(mock_cosmos_container, mocker):
    """Mock the async wrapper functions."""
    async def mock_upsert(doc, partition_key):
        return mock_cosmos_container.upsert_item(doc)
    
    async def mock_get(doc_id, partition_key):
        return mock_cosmos_container.read_item(item=doc_id, partition_key=partition_key)
    
    async def mock_delete(doc_id, partition_key):
        mock_cosmos_container.delete_item(item=doc_id, partition_key=partition_key)
        return True
    
    async def mock_query(doc_type, partition_key=None, extra_filter=None, parameters=None):
        return list(mock_cosmos_container.query_items())
    
    mocker.patch("app.db.cosmos.upsert_document", side_effect=mock_upsert)
    mocker.patch("app.db.cosmos.get_document", side_effect=mock_get)
    mocker.patch("app.db.cosmos.delete_document", side_effect=mock_delete)
    mocker.patch("app.db.cosmos.query_documents", side_effect=mock_query)
    mocker.patch("app.db.cosmos.get_container", return_value=mock_cosmos_container)
    
    return mock_cosmos_container

Unit Test Patterns

Testing Create Operations

class TestProjectServiceCreate:
    
    @pytest.mark.asyncio
    async def test_create_persists_document_with_correct_structure(
        self, mock_cosmos_async, project_create_data
    ):
        # Act
        result = await project_service.create(project_create_data, author_id="user-1")
        
        # Assert - verify upsert was called
        mock_cosmos_async.upsert_item.assert_called_once()
        persisted_doc = mock_cosmos_async.upsert_item.call_args[0][0]
        
        assert persisted_doc["name"] == "New Project"
        assert persisted_doc["workspaceId"] == "ws-456"
        assert persisted_doc["authorId"] == "user-1"
        assert persisted_doc["docType"] == "project"
    
    @pytest.mark.asyncio
    async def test_create_generates_uuid_id(self, mock_cosmos_async, project_create_data):
        result = await project_service.create(project_create_data, author_id="user-1")
        
        # UUID format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
        assert len(result.id) == 36
        assert result.id.count("-") == 4
    
    @pytest.mark.asyncio
    async def test_create_sets_created_at_to_now(self, mock_cosmos_async, project_create_data):
        before = datetime.now(timezone.utc)
        result = await project_service.create(project_create_data, author_id="user-1")
        after = datetime.now(timezone.utc)
        
        assert before <= result.created_at <= after

Testing Read Operations

class TestProjectServiceGetById:
    
    @pytest.mark.asyncio
    async def test_get_by_id_returns_project_when_found(
        self, mock_cosmos_async, sample_project_doc
    ):
        mock_cosmos_async.read_item.return_value = sample_project_doc
        
        result = await project_service.get_by_id("proj-123", workspace_id="ws-456")
        
        assert result is not None
        assert result.id == "proj-123"
        assert result.name == "Test Project"
    
    @pytest.mark.asyncio
    async def test_get_by_id_returns_none_when_not_found(self, mock_cosmos_async):
        from azure.cosmos.exceptions import CosmosResourceNotFoundError
        mock_cosmos_async.read_item.side_effect = CosmosResourceNotFoundError(
            status_code=404, message="Not found"
        )
        
        result = await project_service.get_by_id("nonexistent", workspace_id="ws-456")
        
        assert result is None
    
    @pytest.mark.asyncio
    async def test_get_by_id_returns_none_when_cosmos_unavailable(
        self, mock_cosmos_unavailable
    ):
        result = await project_service.get_by_id("proj-123", workspace_id="ws-456")
        
        assert result is None

Testing Update Operations

class TestProjectServiceUpdate:
    
    @pytest.mark.asyncio
    async def test_update_modifies_only_provided_fields(
        self, mock_cosmos_async, sample_project_doc
    ):
        mock_cosmos_async.read_item.return_value = sample_project_doc
        
        update_data = ProjectUpdate(name="Updated Name")
        result = await project_service.update("proj-123", "ws-456", update_data)
        
        assert result.name == "Updated Name"
        assert result.description == "A test project"  # Unchanged
    
    @pytest.mark.asyncio
    async def test_update_sets_updated_at(self, mock_cosmos_async, sample_project_doc):
        mock_cosmos_async.read_item.return_value = sample_project_doc
        
        before = datetime.now(timezone.utc)
        result = await project_service.update(
            "proj-123", "ws-456", ProjectUpdate(name="New")
        )
        after = datetime.now(timezone.utc)
        
        assert result.updated_at is not None
        assert before <= result.updated_at <= after

Testing Delete Operations

class TestProjectServiceDelete:
    
    @pytest.mark.asyncio
    async def test_delete_returns_true_on_success(self, mock_cosmos_async):
        result = await project_service.delete("proj-123", workspace_id="ws-456")
        
        assert result is True
        mock_cosmos_async.delete_item.assert_called_once()
    
    @pytest.mark.asyncio
    async def test_delete_returns_false_when_not_found(self, mock_cosmos_async):
        from azure.cosmos.exceptions import CosmosResourceNotFoundError
        mock_cosmos_async.delete_item.side_effect = CosmosResourceNotFoundError(
            status_code=404, message="Not found"
        )
        
        result = await project_service.delete("nonexistent", workspace_id="ws-456")
        
        assert result is False

Testing List/Query Operations

class TestProjectServiceList:
    
    @pytest.mark.asyncio
    async def test_list_returns_all_projects_in_workspace(
        self, mock_cosmos_async, sample_project_doc
    ):
        mock_cosmos_async.query_items.return_value = iter([
            sample_project_doc,
            {**sample_project_doc, "id": "proj-456", "name": "Second Project"},
        ])
        
        results = await project_service.list_by_workspace("ws-456")
        
        assert len(results) == 2
        assert results[0].id == "proj-123"
        assert results[1].id == "proj-456"
    
    @pytest.mark.asyncio
    async def test_list_returns_empty_when_no_projects(self, mock_cosmos_async):
        mock_cosmos_async.query_items.return_value = iter([])
        
        results = await project_service.list_by_workspace("ws-456")
        
        assert results == []

Integration Test Patterns

For tests against real Cosmos (emulator or dev instance):

import pytest
import os

# Skip if no emulator configured
pytestmark = pytest.mark.skipif(
    os.getenv("COSMOS_ENDPOINT") is None,
    reason="COSMOS_ENDPOINT not configured"
)


@pytest.fixture(scope="module")
def cosmos_container():
    """Real Cosmos container for integration tests."""
    from app.db.cosmos import get_container, reset_connection
    reset_connection()
    container = get_container()
    yield container
    reset_connection()


@pytest.fixture
def cleanup_test_docs(cosmos_container):
    """Clean up test documents after each test."""
    created_ids = []
    yield created_ids
    for doc_id, partition_key in created_ids:
        try:
            cosmos_container.delete_item(item=doc_id, partition_key=partition_key)
        except Exception:
            pass


class TestProjectServiceIntegration:
    
    @pytest.mark.asyncio
    async def test_create_and_retrieve_roundtrip(self, cleanup_test_docs):
        # Create
        data = ProjectCreate(
            name="Integration Test Project",
            workspace_id="test-workspace",
        )
        project = await project_service.create(data, author_id="test-user")
        cleanup_test_docs.append((project.id, "test-workspace"))
        
        # Retrieve
        retrieved = await project_service.get_by_id(project.id, "test-workspace")
        
        assert retrieved is not None
        assert retrieved.name == "Integration Test Project"

Test Data Factories

For complex test scenarios, use factories:

from dataclasses import dataclass, field
from datetime import datetime, timezone
import uuid


@dataclass
class ProjectFactory:
    """Factory for creating test project data."""
    
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = "Test Project"
    description: str = "Test description"
    slug: str = "test-project"
    workspace_id: str = "ws-test"
    author_id: str = "user-test"
    visibility: str = "public"
    tags: list = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    
    def to_doc(self) -> dict:
        """Convert to Cosmos document format."""
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "slug": self.slug,
            "workspaceId": self.workspace_id,
            "authorId": self.author_id,
            "visibility": self.visibility,
            "tags": self.tags,
            "createdAt": self.created_at.isoformat(),
            "updatedAt": None,
            "docType": "project",
        }
    
    def to_create(self) -> ProjectCreate:
        """Convert to ProjectCreate model."""
        return ProjectCreate(
            name=self.name,
            description=self.description,
            workspace_id=self.workspace_id,
            visibility=self.visibility,
            tags=self.tags,
        )


# Usage in tests
def test_with_factory(mock_cosmos_async):
    factory = ProjectFactory(name="Custom Name", tags=["important"])
    mock_cosmos_async.read_item.return_value = factory.to_doc()
    
    result = await project_service.get_by_id(factory.id, factory.workspace_id)
    assert result.name == "Custom Name"

Complete Fixture File

See assets/conftest_template.py for a production-ready conftest.py.

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs an API key, Needs Python

Security

What our scanner noticed· Low risk

Instructions that use the network
Scripts that your agent may run

This skill includes scripts that your agent may run. Read them before you use it.

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by

Microsoft

Publisher

Microsoft (Official)

Source

microsoft/skills · .github/plugins/azure-sdk-python/skills/azure-cosmos-db-py · commit 0696600

Licence

MIT License (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-09

AutoGineer is not affiliated with or endorsed by Microsoft or Microsoft.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-cosmos-db-py.zip

From the author

AutoGineer summary

What it does

When to use it

Advantages

Usage examples

Install

Documentation by Microsoft

Cosmos DB Service Implementation

Installation

Environment Variables

Authentication & Lifecycle

Architecture Overview

Quick Start

1. Client Module Setup

2. Pydantic Model Hierarchy

3. Service Layer Pattern

Core Principles

Security Requirements

Clean Code Conventions

TDD Requirements

Best Practices

Reference Files

Template Files

Quality Attributes (NFRs)

Reliability

Security

Maintainability

Testability

Performance

Files

references/testing.md

Testing Cosmos DB Services

Table of Contents

Test-Driven Development Workflow

Example TDD Cycle

Test File Structure

Test File Template

Fixtures and Mocking

Core Fixtures (conftest.py)

Mocking Async Operations

Unit Test Patterns

Testing Create Operations

Testing Read Operations

Testing Update Operations

Testing Delete Operations

Testing List/Query Operations

Integration Test Patterns

Test Data Factories

Complete Fixture File

Requirements and compatibility

Security

Attribution and licence

Related skills

Implement Spec

MCP Builder

Schema

Postgresql Table Design

SQL Optimization Patterns

API Scaffolding Fastapi Pro

Download ZIP