# SkillPatch skill: azure-cosmos-db-py

This skill guides agents in building production-grade Azure Cosmos DB NoSQL services using Python and FastAPI. It covers dual authentication strategies (DefaultAzureCredential and emulator), environment configuration, client lifecycle management, service layer architecture, and TDD patterns. It is designed to help implement clean, secure, and testable Cosmos DB integrations from setup through CRUD operations.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/azure-cosmos-db-py
curl -sSL https://skillpatch.dev/install_skill/azure-cosmos-db-py | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (9)

- `SKILL.md`
- `assets/conftest_template.py`
- `assets/cosmos_client_template.py`
- `assets/service_template.py`
- `references/client-setup.md`
- `references/error-handling.md`
- `references/partitioning.md`
- `references/service-layer.md`
- `references/testing.md`


### `SKILL.md`

````markdown
---
name: azure-cosmos-db-py
description: 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".
license: MIT
metadata:
  author: Microsoft
  version: "1.0.0"
  package: azure-cosmos
---

# Cosmos DB Service Implementation

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

## Installation

```bash
pip install azure-cosmos azure-identity
```

## Environment Variables

```bash
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:**
>
> 1. **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.
> 2. **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)**:
```python
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)**:
```python
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:

```python
# 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](references/client-setup.md)

### 2. Pydantic Model Hierarchy

Use five-tier model pattern for clean separation:

```python
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

```python
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](references/service-layer.md)

## Core Principles

### Security Requirements

1. **RBAC Authentication**: Use `DefaultAzureCredential` in Azure — never store keys in code
2. **Emulator-Only Keys**: Hardcode the well-known emulator key only for local development
3. **Parameterized Queries**: Always use `@parameter` syntax — never string concatenation
4. **Partition Key Validation**: Validate partition key access matches user authorization

### Clean Code Conventions

1. **Single Responsibility**: Client module handles connection; services handle business logic
2. **Graceful Degradation**: Services return `None`/`[]` when Cosmos unavailable
3. **Consistent Naming**: `_doc_to_model()`, `_model_to_doc()`, `_use_cosmos()`
4. **Type Hints**: Full typing on all public methods
5. **CamelCase Aliases**: Use `Field(alias="camelCase")` for JSON serialization

### TDD Requirements

Write tests BEFORE implementation using these patterns:

```python
@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](references/testing.md)

## Best Practices

1. **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.
2. **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](references/client-setup.md) | Setting up Cosmos client with dual auth, SSL config, singleton pattern |
| [references/service-layer.md](references/service-layer.md) | Implementing full service class with CRUD, conversions, graceful degradation |
| [references/testing.md](references/testing.md) | Writing pytest tests, mocking Cosmos, integration test setup |
| [references/partitioning.md](references/partitioning.md) | Choosing partition keys, cross-partition queries, move operations |
| [references/error-handling.md](references/error-handling.md) | Handling CosmosResourceNotFoundError, logging, HTTP error mapping |

## Template Files

| File | Purpose |
|------|---------|
| [assets/cosmos_client_template.py](assets/cosmos_client_template.py) | Ready-to-use client module |
| [assets/service_template.py](assets/service_template.py) | Service class skeleton |
| [assets/conftest_template.py](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

````


### `assets/conftest_template.py`

```
"""
pytest Fixtures Template for Cosmos DB Testing

Production-ready test fixtures with:
- Mock Cosmos container
- Async operation mocking
- Sample document factories
- Integration test support

Usage:
    Place in tests/conftest.py or import into test modules.
"""
from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Generator
from unittest.mock import MagicMock

import pytest


# -----------------------------------------------------------------------------
# Core Cosmos Fixtures
# -----------------------------------------------------------------------------


@pytest.fixture
def mock_cosmos_container() -> MagicMock:
    """
    Mock Cosmos container with default behaviors.

    Returns:
        MagicMock configured for common Cosmos operations
    """
    container = MagicMock()

    # Default return values
    container.read_item.return_value = None
    container.upsert_item.side_effect = lambda doc: doc  # Return the doc
    container.delete_item.return_value = None
    container.query_items.return_value = iter([])

    return container


@pytest.fixture
def mock_cosmos(mock_cosmos_container: MagicMock, mocker) -> MagicMock:
    """
    Patch get_container to return mock container.

    Usage:
        def test_something(mock_cosmos):
            mock_cosmos.read_item.return_value = {"id": "123", ...}
            result = await my_service.get_by_id("123", "workspace")
    """
    mocker.patch("app.db.cosmos.get_container", return_value=mock_cosmos_container)
    return mock_cosmos_container


@pytest.fixture
def mock_cosmos_unavailable(mocker) -> None:
    """
    Simulate Cosmos being unavailable.

    Usage:
        def test_graceful_degradation(mock_cosmos_unavailable):
            result = await my_service.list_items("workspace")
            assert result == []
    """
    mocker.patch("app.db.cosmos.get_container", return_value=None)


@pytest.fixture
def mock_cosmos_async(mock_cosmos_container: MagicMock, mocker) -> MagicMock:
    """
    Mock the async wrapper functions (upsert_document, get_document, etc.).

    This is useful when testing service layer directly.
    """
    from azure.cosmos.exceptions import CosmosResourceNotFoundError

    async def mock_upsert(doc: dict, partition_key: str) -> dict:
        return mock_cosmos_container.upsert_item(doc)

    async def mock_get(doc_id: str, partition_key: str) -> dict | None:
        try:
            return mock_cosmos_container.read_item(
                item=doc_id, partition_key=partition_key
            )
        except CosmosResourceNotFoundError:
            return None

    async def mock_delete(doc_id: str, partition_key: str) -> bool:
        try:
            mock_cosmos_container.delete_item(
                item=doc_id, partition_key=partition_key
            )
            return True
        except CosmosResourceNotFoundError:
            return False

    async def mock_query(
        doc_type: str,
        partition_key: str | None = None,
        extra_filter: str | None = None,
        parameters: list | None = None,
    ) -> list[dict]:
        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


# -----------------------------------------------------------------------------
# Test Data Factories
# -----------------------------------------------------------------------------


@dataclass
class ProjectFactory:
    """
    Factory for creating test project data.

    Usage:
        factory = ProjectFactory(name="Custom Name")
        mock_cosmos.read_item.return_value = factory.to_doc()
    """

    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[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime | None = None

    def to_doc(self) -> dict[str, Any]:
        """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": self.updated_at.isoformat() if self.updated_at else None,
            "docType": "project",
        }


@dataclass
class WorkspaceFactory:
    """Factory for creating test workspace data."""

    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = "Test Workspace"
    description: str = "Test workspace description"
    slug: str = "test-workspace"
    owner_id: str = "user-test"
    visibility: str = "private"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def to_doc(self) -> dict[str, Any]:
        """Convert to Cosmos document format."""
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "slug": self.slug,
            "ownerId": self.owner_id,
            "visibility": self.visibility,
            "createdAt": self.created_at.isoformat(),
            "docType": "workspace",
        }


@dataclass
class UserFactory:
    """Factory for creating test user data."""

    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    email: str = "test@example.com"
    name: str = "Test User"
    avatar_url: str | None = None
    role: str = "author"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def to_doc(self) -> dict[str, Any]:
        """Convert to Cosmos document format."""
        return {
            "id": self.id,
            "email": self.email,
            "name": self.name,
            "avatarUrl": self.avatar_url,
            "role": self.role,
            "createdAt": self.created_at.isoformat(),
            "docType": "user",
        }


# -----------------------------------------------------------------------------
# Sample Document Fixtures
# -----------------------------------------------------------------------------


@pytest.fixture
def sample_project_doc() -> dict[str, Any]:
    """Sample project document as stored in Cosmos."""
    return ProjectFactory().to_doc()


@pytest.fixture
def sample_workspace_doc() -> dict[str, Any]:
    """Sample workspace document as stored in Cosmos."""
    return WorkspaceFactory().to_doc()


@pytest.fixture
def sample_user_doc() -> dict[str, Any]:
    """Sample user document as stored in Cosmos."""
    return UserFactory().to_doc()


# -----------------------------------------------------------------------------
# Integration Test Fixtures
# -----------------------------------------------------------------------------


@pytest.fixture(scope="module")
def cosmos_container_integration():
    """
    Real Cosmos container for integration tests.

    Requires COSMOS_ENDPOINT environment variable.
    Skipped if not configured.
    """
    import os

    if not os.getenv("COSMOS_ENDPOINT"):
        pytest.skip("COSMOS_ENDPOINT not configured for integration tests")

    from app.db.cosmos import get_container, reset_connection

    reset_connection()
    container = get_container()

    if container is None:
        pytest.skip("Could not connect to Cosmos DB")

    yield container

    reset_connection()


@pytest.fixture
def cleanup_test_docs(
    cosmos_container_integration,
) -> Generator[list[tuple[str, str]], None, None]:
    """
    Track and clean up test documents after each integration test.

    Usage:
        def test_create(cleanup_test_docs):
            project = await service.create(...)
            cleanup_test_docs.append((project.id, project.workspace_id))
            # Document will be deleted after test
    """
    created: list[tuple[str, str]] = []
    yield created

    # Cleanup
    for doc_id, partition_key in created:
        try:
            cosmos_container_integration.delete_item(
                item=doc_id, partition_key=partition_key
            )
        except Exception:
            pass  # Ignore cleanup errors


# -----------------------------------------------------------------------------
# Model Fixtures (customize for your models)
# -----------------------------------------------------------------------------


@pytest.fixture
def project_create_data():
    """
    Sample ProjectCreate for testing.

    TODO: Import and use your actual ProjectCreate model.
    """
    # from app.models.project import ProjectCreate
    # return ProjectCreate(
    #     name="New Project",
    #     description="Test description",
    #     workspace_id="ws-test",
    #     visibility="public",
    #     tags=["test"],
    # )
    return {
        "name": "New Project",
        "description": "Test description",
        "workspace_id": "ws-test",
        "visibility": "public",
        "tags": ["test"],
    }


@pytest.fixture
def project_update_data():
    """
    Sample ProjectUpdate for testing.

    TODO: Import and use your actual ProjectUpdate model.
    """
    # from app.models.project import ProjectUpdate
    # return ProjectUpdate(name="Updated Name")
    return {"name": "Updated Name"}

```


### `assets/cosmos_client_template.py`

```
"""
Cosmos DB Client Module Template

Production-ready Azure Cosmos DB NoSQL client with:
- Dual authentication (DefaultAzureCredential for Azure, key for emulator)
- Singleton pattern for connection reuse
- Async wrapping via run_in_threadpool
- Graceful error handling

Usage:
    from app.db.cosmos import get_container, upsert_document, get_document
"""
from __future__ import annotations

import logging
from typing import Any, Optional

from azure.cosmos import ContainerProxy, CosmosClient
from azure.cosmos.exceptions import CosmosResourceNotFoundError
from azure.identity import DefaultAzureCredential
from starlette.concurrency import run_in_threadpool

from app.config import settings

logger = logging.getLogger(__name__)

# Module-level singleton state
_cosmos_container: Optional[ContainerProxy] = None
_credential: Optional[DefaultAzureCredential] = None
_init_attempted: bool = False


def _is_emulator_endpoint(endpoint: str) -> bool:
    """Detect if endpoint is Cosmos emulator."""
    return "localhost" in endpoint.lower() or "127.0.0.1" in endpoint


def _create_client() -> CosmosClient:
    """Create Cosmos client with appropriate authentication."""
    global _credential

    if _is_emulator_endpoint(settings.cosmos_endpoint):
        logger.info("Using Cosmos emulator with key authentication")
        return CosmosClient(
            url=settings.cosmos_endpoint,
            credential=settings.cosmos_key,
            connection_verify=False,  # Emulator uses self-signed cert
        )
    else:
        logger.info("Using Azure Cosmos with DefaultAzureCredential (RBAC)")
        _credential = DefaultAzureCredential()
        return CosmosClient(
            url=settings.cosmos_endpoint,
            credential=_credential,
        )


def get_container() -> Optional[ContainerProxy]:
    """
    Get Cosmos container proxy, initializing on first call.

    Returns:
        ContainerProxy if connection successful, None otherwise.
    """
    global _cosmos_container, _init_attempted

    if _init_attempted:
        return _cosmos_container

    _init_attempted = True

    try:
        client = _create_client()
        database = client.get_database_client(settings.cosmos_database_name)
        _cosmos_container = database.get_container_client(settings.cosmos_container_id)

        # Verify connection with lightweight operation
        _cosmos_container.read()

        logger.info(
            f"✅ Cosmos DB connected: {settings.cosmos_database_name}/{settings.cosmos_container_id}"
        )

    except Exception as e:
        logger.error(f"❌ Cosmos DB connection failed: {type(e).__name__}: {e}")
        import traceback

        logger.error(traceback.format_exc())
        _cosmos_container = None

    return _cosmos_container


def reset_connection() -> None:
    """Reset connection for testing or environment switching."""
    global _cosmos_container, _credential, _init_attempted
    _cosmos_container = None
    _credential = None
    _init_attempted = False


# -----------------------------------------------------------------------------
# Async CRUD Operations
# -----------------------------------------------------------------------------


async def upsert_document(doc: dict[str, Any], partition_key: str) -> dict[str, Any]:
    """
    Insert or update a document.

    Args:
        doc: Document to upsert (must include 'id' field)
        partition_key: Partition key value

    Returns:
        The upserted document

    Raises:
        RuntimeError: If Cosmos is not initialized
    """
    container = get_container()
    if container is None:
        raise RuntimeError("Cosmos DB not initialized")

    result = await run_in_threadpool(container.upsert_item, doc)
    return result


async def get_document(doc_id: str, partition_key: str) -> Optional[dict[str, Any]]:
    """
    Read a document by ID.

    Args:
        doc_id: Document ID
        partition_key: Partition key value

    Returns:
        Document dict if found, None otherwise
    """
    container = get_container()
    if container is None:
        return None

    try:
        result = await run_in_threadpool(
            container.read_item,
            item=doc_id,
            partition_key=partition_key,
        )
        return result
    except CosmosResourceNotFoundError:
        return None


async def delete_document(doc_id: str, partition_key: str) -> bool:
    """
    Delete a document.

    Args:
        doc_id: Document ID
        partition_key: Partition key value

    Returns:
        True if deleted, False if not found
    """
    container = get_container()
    if container is None:
        return False

    try:
        await run_in_threadpool(
            container.delete_item,
            item=doc_id,
            partition_key=partition_key,
        )
        return True
    except CosmosResourceNotFoundError:
        return False


async def query_documents(
    doc_type: str,
    partition_key: Optional[str] = None,
    extra_filter: Optional[str] = None,
    parameters: Optional[list[dict[str, Any]]] = None,
) -> list[dict[str, Any]]:
    """
    Query documents by type with optional filters.

    Args:
        doc_type: Document type to filter by (docType field)
        partition_key: Partition key for efficient query (None for cross-partition)
        extra_filter: Additional SQL WHERE clause (e.g., "AND c.slug = @slug")
        parameters: Query parameters list (e.g., [{"name": "@slug", "value": "my-slug"}])

    Returns:
        List of matching documents
    """
    container = get_container()
    if container is None:
        return []

    # Build parameterized query
    query = "SELECT * FROM c WHERE c.docType = @docType"
    query_params: list[dict[str, Any]] = [{"name": "@docType", "value": doc_type}]

    if extra_filter:
        query += f" {extra_filter}"
        query_params.extend(parameters or [])

    # Execute query
    if partition_key:
        items = container.query_items(
            query=query,
            parameters=query_params,
            partition_key=partition_key,
        )
    else:
        items = container.query_items(
            query=query,
            parameters=query_params,
            enable_cross_partition_query=True,
        )

    return await run_in_threadpool(list, items)

```


### `assets/service_template.py`

```
"""
Service Layer Template

Production-ready service class pattern for Cosmos DB with:
- Document ↔ Model conversion
- CRUD operations
- Graceful degradation
- Unique slug generation

Usage:
    Rename and customize for your entity type.
"""
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any, Optional

from app.db.cosmos import (
    delete_document,
    get_container,
    get_document,
    query_documents,
    upsert_document,
)

# TODO: Import your Pydantic models
# from app.models.entity import Entity, EntityCreate, EntityUpdate, EntityInDB


def slugify(text: str) -> str:
    """Convert text to URL-friendly slug."""
    import re

    slug = text.lower().strip()
    slug = re.sub(r"[^\w\s-]", "", slug)
    slug = re.sub(r"[\s_]+", "-", slug)
    slug = re.sub(r"-+", "-", slug)
    return slug.strip("-")


class EntityService:
    """
    Service for Entity CRUD operations.

    Replace 'Entity' with your actual entity name (Project, Workspace, etc.)
    """

    # -------------------------------------------------------------------------
    # Helper Methods
    # -------------------------------------------------------------------------

    def _use_cosmos(self) -> bool:
        """Check if Cosmos DB is available."""
        return get_container() is not None

    def _doc_to_model_in_db(self, doc: dict[str, Any]) -> "EntityInDB":
        """
        Convert Cosmos document to internal model.

        Maps camelCase JSON fields to snake_case Python attributes.
        """
        # TODO: Customize for your entity
        return EntityInDB(
            id=doc["id"],
            name=doc["name"],
            description=doc.get("description"),
            slug=doc["slug"],
            workspace_id=doc["workspaceId"],  # Partition key
            author_id=doc["authorId"],
            visibility=doc.get("visibility", "public"),
            tags=doc.get("tags", []),
            created_at=datetime.fromisoformat(doc["createdAt"]),
            updated_at=(
                datetime.fromisoformat(doc["updatedAt"]) if doc.get("updatedAt") else None
            ),
            doc_type=doc.get("docType", "entity"),
        )

    def _model_in_db_to_doc(self, model: "EntityInDB") -> dict[str, Any]:
        """
        Convert internal model to Cosmos document.

        Maps snake_case Python attributes to camelCase JSON fields.
        """
        # TODO: Customize for your entity
        return {
            "id": model.id,
            "name": model.name,
            "description": model.description,
            "slug": model.slug,
            "workspaceId": model.workspace_id,  # Partition key
            "authorId": model.author_id,
            "visibility": model.visibility,
            "tags": model.tags,
            "createdAt": model.created_at.isoformat(),
            "updatedAt": model.updated_at.isoformat() if model.updated_at else None,
            "docType": model.doc_type,
        }

    def _model_in_db_to_model(self, model_in_db: "EntityInDB") -> "Entity":
        """
        Convert internal model to API response model.

        Strips internal fields like doc_type.
        """
        # TODO: Customize for your entity
        return Entity(
            id=model_in_db.id,
            name=model_in_db.name,
            description=model_in_db.description,
            slug=model_in_db.slug,
            workspace_id=model_in_db.workspace_id,
            author_id=model_in_db.author_id,
            visibility=model_in_db.visibility,
            tags=model_in_db.tags,
            created_at=model_in_db.created_at,
            updated_at=model_in_db.updated_at,
        )

    async def _generate_unique_slug(self, name: str, workspace_id: str) -> str:
        """Generate unique slug within workspace."""
        base_slug = slugify(name)
        slug = base_slug
        counter = 1

        while True:
            existing = await query_documents(
                doc_type="entity",  # TODO: Change to your entity type
                partition_key=workspace_id,
                extra_filter="AND c.slug = @slug",
                parameters=[{"name": "@slug", "value": slug}],
            )
            if not existing:
                return slug
            slug = f"{base_slug}-{counter}"
            counter += 1

    # -------------------------------------------------------------------------
    # CRUD Operations
    # -------------------------------------------------------------------------

    async def create(self, data: "EntityCreate", author_id: str) -> "Entity":
        """
        Create a new entity.

        Args:
            data: Creation request data
            author_id: ID of the creating user

        Returns:
            Created entity

        Raises:
            RuntimeError: If Cosmos is unavailable
        """
        if not self._use_cosmos():
            raise RuntimeError("Database unavailable")

        now = datetime.now(timezone.utc)
        slug = await self._generate_unique_slug(data.name, data.workspace_id)

        entity_in_db = EntityInDB(
            id=str(uuid.uuid4()),
            name=data.name,
            description=data.description,
            slug=slug,
            workspace_id=data.workspace_id,
            author_id=author_id,
            visibility=data.visibility,
            tags=data.tags or [],
            created_at=now,
            updated_at=None,
            doc_type="entity",  # TODO: Change to your entity type
        )

        doc = self._model_in_db_to_doc(entity_in_db)
        await upsert_document(doc, partition_key=data.workspace_id)

        return self._model_in_db_to_model(entity_in_db)

    async def get_by_id(
        self, entity_id: str, workspace_id: str
    ) -> Optional["Entity"]:
        """
        Get entity by ID.

        Args:
            entity_id: Entity ID
            workspace_id: Workspace ID (partition key)

        Returns:
            Entity if found, None otherwise
        """
        if not self._use_cosmos():
            return None

        doc = await get_document(entity_id, partition_key=workspace_id)
        if doc is None:
            return None

        model_in_db = self._doc_to_model_in_db(doc)
        return self._model_in_db_to_model(model_in_db)

    async def get_by_slug(
        self, slug: str, workspace_id: str
    ) -> Optional["Entity"]:
        """
        Get entity by slug within a workspace.

        Args:
            slug: URL-friendly slug
            workspace_id: Workspace ID (partition key)

        Returns:
            Entity if found, None otherwise
        """
        if not self._use_cosmos():
            return None

        docs = await query_documents(
            doc_type="entity",  # TODO: Change to your entity type
            partition_key=workspace_id,
            extra_filter="AND c.slug = @slug",
            parameters=[{"name": "@slug", "value": slug}],
        )

        if not docs:
            return None

        model_in_db = self._doc_to_model_in_db(docs[0])
        return self._model_in_db_to_model(model_in_db)

    async def update(
        self,
        entity_id: str,
        workspace_id: str,
        data: "EntityUpdate",
    ) -> Optional["Entity"]:
        """
        Update entity.

        Args:
            entity_id: Entity ID
            workspace_id: Workspace ID (partition key)
            data: Update data (only non-None fields are applied)

        Returns:
            Updated entity if found, None otherwise
        """
        if not self._use_cosmos():
            return None

        doc = await get_document(entity_id, partition_key=workspace_id)
        if doc is None:
            return None

        model_in_db = self._doc_to_model_in_db(doc)

        # Apply updates (only non-None fields)
        update_data = data.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            if hasattr(model_in_db, field):
                setattr(model_in_db, field, value)

        model_in_db.updated_at = datetime.now(timezone.utc)

        updated_doc = self._model_in_db_to_doc(model_in_db)
        await upsert_document(updated_doc, partition_key=workspace_id)

        return self._model_in_db_to_model(model_in_db)

    async def delete(self, entity_id: str, workspace_id: str) -> bool:
        """
        Delete entity.

        Args:
            entity_id: Entity ID
            workspace_id: Workspace ID (partition key)

        Returns:
            True if deleted, False if not found
        """
        if not self._use_cosmos():
            return False

        return await delete_document(entity_id, partition_key=workspace_id)

    async def list_by_workspace(self, workspace_id: str) -> list["Entity"]:
        """
        List all entities in a workspace.

        Args:
            workspace_id: Workspace ID (partition key)

        Returns:
            List of entities (empty if unavailable)
        """
        if not self._use_cosmos():
            return []

        docs = await query_documents(
            doc_type="entity",  # TODO: Change to your entity type
            partition_key=workspace_id,
        )

        return [
            self._model_in_db_to_model(self._doc_to_model_in_db(doc))
            for doc in docs
        ]


# Singleton instance
entity_service = EntityService()

```


### `references/client-setup.md`

````markdown
# Cosmos DB Client Setup

## Table of Contents

1. [Dual Authentication Strategy](#dual-authentication-strategy)
2. [Singleton Pattern](#singleton-pattern)
3. [Async Wrapping](#async-wrapping)
4. [Configuration Management](#configuration-management)
5. [Connection Reset](#connection-reset)
6. [Complete Implementation](#complete-implementation)

---

## Dual Authentication Strategy

Use `DefaultAzureCredential` for Azure deployments and key-based auth only for the local emulator:

```python
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential

def _is_emulator_endpoint(endpoint: str) -> bool:
    """Detect Cosmos emulator by endpoint URL."""
    return "localhost" in endpoint.lower() or "127.0.0.1" in endpoint

def _create_client(settings) -> CosmosClient:
    if _is_emulator_endpoint(settings.cosmos_endpoint):
        # Emulator: use well-known key, disable SSL verification
        return CosmosClient(
            url=settings.cosmos_endpoint,
            credential=settings.cosmos_key,
            connection_verify=False  # Emulator uses self-signed cert
        )
    else:
        # Azure: use RBAC via DefaultAzureCredential
        credential = DefaultAzureCredential()
        return CosmosClient(
            url=settings.cosmos_endpoint,
            credential=credential
        )
```

### Security Notes

- **Never** store production keys in code or config files
- The emulator key (`C2y6yDjf5/R+ob0N8A7Cgv...`) is publicly documented and safe to hardcode
- Requires "Cosmos DB Built-in Data Contributor" role for RBAC auth
- Run `az login` before local development against Azure Cosmos

---

## Singleton Pattern

Initialize container once at module level to reuse connections:

```python
import logging
from typing import Optional
from azure.cosmos import ContainerProxy

logger = logging.getLogger(__name__)

_cosmos_container: Optional[ContainerProxy] = None
_init_attempted: bool = False

def get_container() -> Optional[ContainerProxy]:
    """Get Cosmos container, initializing on first call."""
    global _cosmos_container, _init_attempted
    
    if _init_attempted:
        return _cosmos_container
    
    _init_attempted = True
    
    try:
        client = _create_client(settings)
        database = client.get_database_client(settings.cosmos_database_name)
        _cosmos_container = database.get_container_client(settings.cosmos_container_id)
        
        # Verify connection with a lightweight operation
        _cosmos_container.read()
        logger.info(f"✅ Connected to Cosmos DB: {settings.cosmos_database_name}/{settings.cosmos_container_id}")
        
    except Exception as e:
        logger.error(f"❌ Failed to initialize Cosmos DB: {type(e).__name__}: {e}")
        import traceback
        logger.error(traceback.format_exc())
        _cosmos_container = None
    
    return _cosmos_container
```

### Why Singleton?

1. **Connection Reuse**: Cosmos SDK manages connection pooling internally
2. **Startup Validation**: Fail fast if Cosmos is misconfigured
3. **Graceful Degradation**: Services check `get_container() is not None`

---

## Async Wrapping

The Cosmos Python SDK is synchronous. Wrap calls with `run_in_threadpool` to avoid blocking FastAPI:

```python
from starlette.concurrency import run_in_threadpool

async def upsert_document(doc: dict, partition_key: str) -> dict:
    """Insert or update a document."""
    container = get_container()
    if container is None:
        raise RuntimeError("Cosmos DB not initialized")
    
    result = await run_in_threadpool(
        container.upsert_item,
        doc
    )
    return result

async def get_document(doc_id: str, partition_key: str) -> Optional[dict]:
    """Read a document by ID."""
    container = get_container()
    if container is None:
        return None
    
    try:
        result = await run_in_threadpool(
            container.read_item,
            item=doc_id,
            partition_key=partition_key
        )
        return result
    except CosmosResourceNotFoundError:
        return None

async def delete_document(doc_id: str, partition_key: str) -> bool:
    """Delete a document. Returns True if deleted."""
    container = get_container()
    if container is None:
        return False
    
    try:
        await run_in_threadpool(
            container.delete_item,
            item=doc_id,
            partition_key=partition_key
        )
        return True
    except CosmosResourceNotFoundError:
        return False
```

---

## Configuration Management

Use Pydantic Settings with environment-aware defaults:

```python
from pydantic_settings import BaseSettings
from pydantic import model_validator

class Settings(BaseSettings):
    environment: str = "local"
    
    cosmos_endpoint: str = ""
    cosmos_key: str = ""  # Only for emulator
    cosmos_database_name: str = "my-database"
    cosmos_container_id: str = "my-container"
    
    class Config:
        env_file = ".env"
        extra = "ignore"
    
    @model_validator(mode="after")
    def configure_for_environment(self) -> "Settings":
        """Set defaults based on environment."""
        if self.environment == "local" and not self.cosmos_endpoint:
            # Default to emulator
            self.cosmos_endpoint = "https://localhost:8081"
            self.cosmos_key = "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4GUg9jsLAi7dnvNdc0SKFO+GRNpsq24UdBkbFCL/6C+iW9zQ=="
        return self

settings = Settings()
```

### Environment Variables

| Variable | Description | Required |
|----------|-------------|----------|
| `COSMOS_ENDPOINT` | Cosmos account URL | Yes |
| `COSMOS_KEY` | Account key (emulator only) | No |
| `COSMOS_DATABASE_NAME` | Database name | Yes |
| `COSMOS_CONTAINER_ID` | Container name | Yes |

---

## Connection Reset

For testing or environment switching:

```python
def reset_connection() -> None:
    """Reset Cosmos connection. Call between tests or after config changes."""
    global _cosmos_container, _init_attempted
    _cosmos_container = None
    _init_attempted = False
```

---

## Complete Implementation

See [assets/cosmos_client_template.py](../assets/cosmos_client_template.py) for a production-ready implementation.

````


### `references/error-handling.md`

````markdown
# Error Handling Patterns

## Table of Contents

1. [Cosmos Exception Types](#cosmos-exception-types)
2. [Client-Level Error Handling](#client-level-error-handling)
3. [Service-Level Error Handling](#service-level-error-handling)
4. [Router-Level Error Mapping](#router-level-error-mapping)
5. [Logging Patterns](#logging-patterns)
6. [Retry Strategies](#retry-strategies)

---

## Cosmos Exception Types

### Common Exceptions

```python
from azure.cosmos.exceptions import (
    CosmosResourceNotFoundError,    # 404 - Document not found
    CosmosResourceExistsError,      # 409 - Conflict (duplicate ID)
    CosmosHttpResponseError,        # Base class for HTTP errors
)
```

| Exception | HTTP Status | Common Cause |
|-----------|-------------|--------------|
| `CosmosResourceNotFoundError` | 404 | Document/container doesn't exist |
| `CosmosResourceExistsError` | 409 | Document with ID already exists |
| `CosmosHttpResponseError` | 429 | Rate limiting (too many RU) |
| `CosmosHttpResponseError` | 503 | Service unavailable |

---

## Client-Level Error Handling

Handle exceptions in the Cosmos client module:

```python
from azure.cosmos.exceptions import CosmosResourceNotFoundError, CosmosHttpResponseError
import logging

logger = logging.getLogger(__name__)


async def get_document(doc_id: str, partition_key: str) -> dict | None:
    """Read a document. Returns None if not found."""
    container = get_container()
    if container is None:
        return None
    
    try:
        result = await run_in_threadpool(
            container.read_item,
            item=doc_id,
            partition_key=partition_key,
        )
        return result
    except CosmosResourceNotFoundError:
        # Expected case: document doesn't exist
        return None
    except CosmosHttpResponseError as e:
        # Log unexpected errors but don't crash
        logger.error(f"Cosmos error reading {doc_id}: {e.status_code} - {e.message}")
        raise


async def delete_document(doc_id: str, partition_key: str) -> bool:
    """Delete a document. Returns True if deleted, False if not found."""
    container = get_container()
    if container is None:
        return False
    
    try:
        await run_in_threadpool(
            container.delete_item,
            item=doc_id,
            partition_key=partition_key,
        )
        return True
    except CosmosResourceNotFoundError:
        # Already deleted or never existed
        return False


async def upsert_document(doc: dict, partition_key: str) -> dict:
    """Insert or update a document."""
    container = get_container()
    if container is None:
        raise RuntimeError("Cosmos DB not initialized")
    
    try:
        result = await run_in_threadpool(container.upsert_item, doc)
        return result
    except CosmosResourceExistsError:
        # Shouldn't happen with upsert, but handle just in case
        logger.warning(f"Unexpected conflict upserting {doc.get('id')}")
        raise
```

### Return Value Conventions

| Scenario | Return Value |
|----------|--------------|
| Document found | `dict` (the document) |
| Document not found | `None` |
| Delete succeeded | `True` |
| Delete target not found | `False` |
| Cosmos unavailable | `None` or `False` (graceful) |
| Unexpected error | Raise exception |

---

## Service-Level Error Handling

Services interpret client results and apply business logic:

```python
class ProjectService:
    
    async def get_by_id(self, project_id: str, workspace_id: str) -> Project | None:
        """Get project. Returns None if not found or unavailable."""
        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)
    
    async def create(self, data: ProjectCreate, author_id: str) -> Project:
        """Create project. Raises RuntimeError if Cosmos unavailable."""
        if not self._use_cosmos():
            raise RuntimeError("Database unavailable")
        
        # ... create logic ...
    
    async def update(
        self, project_id: str, workspace_id: str, data: ProjectUpdate
    ) -> Project | None:
        """Update project. Returns None if not found."""
        if not self._use_cosmos():
            return None
        
        doc = await get_document(project_id, partition_key=workspace_id)
        if doc is None:
            return None  # Not found
        
        # ... update logic ...
```

### Graceful Degradation Pattern

```python
def _use_cosmos(self) -> bool:
    """Check if Cosmos is available for use."""
    return get_container() is not None

async def list_projects(self, workspace_id: str) -> list[Project]:
    """List projects. Returns empty list if unavailable."""
    if not self._use_cosmos():
        return []  # Graceful empty response
    
    docs = await query_documents(
        doc_type="project",
        partition_key=workspace_id,
    )
    return [self._doc_to_model(doc) for doc in docs]
```

---

## Router-Level Error Mapping

Convert service results to HTTP responses:

```python
from fastapi import APIRouter, HTTPException, status, Depends

router = APIRouter(prefix="/api/projects", tags=["projects"])


@router.get("/{project_id}")
async def get_project(
    project_id: str,
    workspace_id: str,
    current_user: User = Depends(get_current_user_required),
) -> Project:
    """Get project by ID."""
    project = await project_service.get_by_id(project_id, workspace_id)
    
    if project is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Project not found",
        )
    
    return project


@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_project(
    data: ProjectCreate,
    current_user: User = Depends(get_current_user_required),
) -> Project:
    """Create a new project."""
    try:
        return await project_service.create(data, author_id=current_user.id)
    except RuntimeError as e:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=str(e),
        )


@router.put("/{project_id}")
async def update_project(
    project_id: str,
    workspace_id: str,
    data: ProjectUpdate,
    current_user: User = Depends(get_current_user_required),
) -> Project:
    """Update project."""
    project = await project_service.update(project_id, workspace_id, data)
    
    if project is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Project not found",
        )
    
    return project


@router.delete("/{project_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_project(
    project_id: str,
    workspace_id: str,
    current_user: User = Depends(get_current_user_required),
) -> None:
    """Delete project."""
    deleted = await project_service.delete(project_id, workspace_id)
    
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Project not found",
        )
```

### HTTP Status Code Mapping

| Service Result | HTTP Status | Response |
|----------------|-------------|----------|
| Success with data | 200 OK | JSON body |
| Created | 201 Created | JSON body |
| Deleted | 204 No Content | Empty |
| Not found (`None`) | 404 Not Found | Error detail |
| Validation error | 422 Unprocessable | Pydantic errors |
| Database unavailable | 503 Service Unavailable | Error detail |

---

## Logging Patterns

### Structured Logging

```python
import logging
import traceback

logger = logging.getLogger(__name__)


def get_container() -> ContainerProxy | None:
    global _cosmos_container, _init_attempted
    
    if _init_attempted:
        return _cosmos_container
    
    _init_attempted = True
    
    try:
        client = _create_client(settings)
        database = client.get_database_client(settings.cosmos_database_name)
        _cosmos_container = database.get_container_client(settings.cosmos_container_id)
        _cosmos_container.read()  # Verify connection
        
        logger.info(
            "Cosmos DB connected",
            extra={
                "database": settings.cosmos_database_name,
                "container": settings.cosmos_container_id,
                "endpoint": settings.cosmos_endpoint[:30] + "...",
            }
        )
        
    except Exception as e:
        logger.error(
            "Cosmos DB connection failed",
            extra={
                "error_type": type(e).__name__,
                "error_message": str(e),
                "endpoint": settings.cosmos_endpoint[:30] + "...",
            },
            exc_info=True,  # Include stack trace
        )
        _cosmos_container = None
    
    return _cosmos_container
```

### Log Levels

| Scenario | Level | Example |
|----------|-------|---------|
| Connection established | INFO | "Cosmos DB connected" |
| Document not found | DEBUG | "Project proj-123 not found" |
| Validation error | WARNING | "Invalid partition key" |
| Connection failed | ERROR | "Cosmos DB connection failed" |
| Unexpected exception | ERROR | Full traceback |

---

## Retry Strategies

### Azure SDK Built-in Retry

The Cosmos SDK has built-in retry for transient errors (429, 503):

```python
from azure.cosmos import CosmosClient

client = CosmosClient(
    url=endpoint,
    credential=credential,
    connection_retry_policy={
        "retry_total": 3,
        "retry_backoff_factor": 0.8,
    }
)
```

### Custom Retry for Application Logic

```python
import asyncio
from typing import TypeVar, Callable

T = TypeVar("T")


async def with_retry(
    operation: Callable[[], T],
    max_retries: int = 3,
    base_delay: float = 0.5,
) -> T:
    """Execute operation with exponential backoff retry."""
    last_error = None
    
    for attempt in range(max_retries + 1):
        try:
            return await operation()
        except CosmosHttpResponseError as e:
            if e.status_code == 429:  # Rate limited
                last_error = e
                delay = base_delay * (2 ** attempt)
                logger.warning(f"Rate limited, retrying in {delay}s (attempt {attempt + 1})")
                await asyncio.sleep(delay)
            else:
                raise
    
    raise last_error


# Usage
result = await with_retry(
    lambda: upsert_document(doc, partition_key),
    max_retries=3,
)
```

````


### `references/partitioning.md`

````markdown
# Partition Key Strategies

## Table of Contents

1. [Partition Key Fundamentals](#partition-key-fundamentals)
2. [Common Strategies](#common-strategies)
3. [Cross-Partition Queries](#cross-partition-queries)
4. [Move Operations](#move-operations)
5. [Query Optimization](#query-optimization)

---

## Partition Key Fundamentals

### What is a Partition Key?

The partition key determines data distribution and query efficiency:

- **Same partition key** → Data co-located → Fast queries, transactional writes
- **Different partition keys** → Data distributed → Cross-partition queries required

### Choosing a Partition Key

Good partition keys have:

1. **High cardinality** — Many distinct values to distribute load
2. **Even distribution** — No single value dominates storage/throughput
3. **Query alignment** — Most queries filter by partition key

---

## Common Strategies

### Strategy 1: Self-Partitioned Entities

Use the entity's own ID when entities are accessed individually:

```python
# Workspaces partition by their own ID
class WorkspaceInDB(BaseModel):
    id: str                    # Also used as partition key
    name: str
    doc_type: str = "workspace"

# Query always includes the workspace ID
doc = await get_document(workspace_id, partition_key=workspace_id)
```

**Use for**: Top-level entities with no parent (workspaces, users)

### Strategy 2: Parent-Scoped Partitioning

Use parent entity ID for child entities accessed together:

```python
# Projects partition by workspace_id
class ProjectInDB(BaseModel):
    id: str
    workspace_id: str          # Partition key
    name: str
    doc_type: str = "project"

# All projects in a workspace are co-located
projects = await query_documents(
    doc_type="project",
    partition_key=workspace_id,  # Efficient single-partition query
)
```

**Use for**: Child entities frequently queried with parent (projects→workspace, flows→workspace)

### Strategy 3: Global Partition

Use a constant for cross-cutting entities:

```python
# Users and groups use "global" partition
GLOBAL_PARTITION = "global"

class UserInDB(BaseModel):
    id: str
    email: str
    doc_type: str = "user"

# Users are queried by email across all partitions
user = await query_documents(
    doc_type="user",
    partition_key=GLOBAL_PARTITION,
    extra_filter="AND c.email = @email",
    parameters=[{"name": "@email", "value": email}],
)
```

**Use for**: Entities accessed independently of hierarchy (users, groups, permissions)

### Strategy Summary

| Entity Type | Partition Key | Rationale |
|-------------|---------------|-----------|
| **Workspace** | `workspace.id` (self) | Independent top-level entity |
| **User** | `"global"` | Accessed by email across workspaces |
| **Group** | `"global"` | Cross-workspace access control |
| **Project** | `workspace_id` | Always accessed within workspace context |
| **Flow** | `workspace_id` | Co-located with parent project |
| **Asset** | `owner_id` | User-scoped media assets |

---

## Cross-Partition Queries

When partition key is unknown, enable cross-partition queries:

```python
async def query_documents(
    doc_type: str,
    partition_key: str | None = None,
    extra_filter: str | None = None,
    parameters: list[dict] | None = None,
) -> list[dict]:
    """Query documents, optionally across partitions."""
    container = get_container()
    
    query = "SELECT * FROM c WHERE c.docType = @docType"
    query_params = [{"name": "@docType", "value": doc_type}]
    
    if extra_filter:
        query += f" {extra_filter}"
        query_params.extend(parameters or [])
    
    if partition_key:
        # Efficient single-partition query
        items = container.query_items(
            query=query,
            parameters=query_params,
            partition_key=partition_key,
        )
    else:
        # Cross-partition query (slower, higher RU cost)
        items = container.query_items(
            query=query,
            parameters=query_params,
            enable_cross_partition_query=True,
        )
    
    return await run_in_threadpool(list, items)
```

### When to Use Cross-Partition Queries

| Scenario | Approach |
|----------|----------|
| Search by email | Cross-partition (email not in partition key) |
| List user's workspaces | Cross-partition (user owns multiple workspaces) |
| Admin dashboard | Cross-partition (aggregate across all) |
| Project by ID in workspace | Single-partition (workspace_id known) |

### Performance Considerations

- **Single-partition**: ~1-5 RU per query
- **Cross-partition**: ~N × single-partition RU (N = physical partitions)
- Index cross-partition query fields for better performance

---

## Move Operations

Moving entities between partitions requires delete + insert:

```python
async def move_project_to_workspace(
    self,
    project_id: str,
    old_workspace_id: str,
    new_workspace_id: str,
) -> Project | None:
    """Move project to a different workspace."""
    # 1. Read from old partition
    doc = await get_document(project_id, partition_key=old_workspace_id)
    if doc is None:
        return None
    
    # 2. Update partition key value
    model_in_db = self._doc_to_model_in_db(doc)
    model_in_db.workspace_id = new_workspace_id
    model_in_db.updated_at = datetime.now(timezone.utc)
    
    # 3. Insert into new partition
    new_doc = self._model_in_db_to_doc(model_in_db)
    await upsert_document(new_doc, partition_key=new_workspace_id)
    
    # 4. Delete from old partition
    await delete_document(project_id, partition_key=old_workspace_id)
    
    return self._model_in_db_to_model(model_in_db)
```

### Atomic Move with Transaction

For critical moves, use transactional batch (same partition only):

```python
# Note: Batch only works within single partition
# For cross-partition moves, implement saga pattern with compensation
```

---

## Query Optimization

### Index Policy Configuration

Ensure frequently queried fields are indexed:

```json
{
  "indexingMode": "consistent",
  "includedPaths": [
    { "path": "/docType/?" },
    { "path": "/workspaceId/?" },
    { "path": "/authorId/?" },
    { "path": "/slug/?" },
    { "path": "/email/?" },
    { "path": "/tags/[]" }
  ],
  "excludedPaths": [
    { "path": "/content/*" },
    { "path": "/nodes/*" }
  ]
}
```

### Composite Indexes

For queries with multiple filters and ORDER BY:

```json
{
  "compositeIndexes": [
    [
      { "path": "/docType", "order": "ascending" },
      { "path": "/createdAt", "order": "descending" }
    ],
    [
      { "path": "/workspaceId", "order": "ascending" },
      { "path": "/docType", "order": "ascending" },
      { "path": "/name", "order": "ascending" }
    ]
  ]
}
```

### Query Patterns

```python
# Efficient: Uses partition key + indexed field
docs = await query_documents(
    doc_type="project",
    partition_key=workspace_id,
    extra_filter="AND c.visibility = @visibility ORDER BY c.createdAt DESC",
    parameters=[{"name": "@visibility", "value": "public"}],
)

# Less efficient: Cross-partition with unindexed field
docs = await query_documents(
    doc_type="project",
    partition_key=None,  # Cross-partition
    extra_filter="AND CONTAINS(c.description, @search)",  # Unindexed scan
    parameters=[{"name": "@search", "value": "keyword"}],
)
```

````


### `references/service-layer.md`

````markdown
# Service Layer Pattern

## Table of Contents

1. [Service Class Structure](#service-class-structure)
2. [Document Conversion Methods](#document-conversion-methods)
3. [CRUD Operations](#crud-operations)
4. [Query Patterns](#query-patterns)
5. [Graceful Degradation](#graceful-degradation)
6. [Complete Example](#complete-example)

---

## Service Class Structure

Every service follows this pattern:

```python
from typing import Optional
from app.db.cosmos import get_container, upsert_document, get_document, delete_document, query_documents
from app.models.project import Project, ProjectCreate, ProjectUpdate, ProjectInDB

class ProjectService:
    """Service for project CRUD operations."""
    
    def _use_cosmos(self) -> bool:
        """Check if Cosmos is available."""
        return get_container() is not None
    
    # Document conversion methods
    def _doc_to_model_in_db(self, doc: dict) -> ProjectInDB:
        """Convert Cosmos document to internal model."""
        ...
    
    def _model_in_db_to_doc(self, model: ProjectInDB) -> dict:
        """Convert internal model to Cosmos document."""
        ...
    
    def _model_in_db_to_model(self, model_in_db: ProjectInDB) -> Project:
        """Convert internal model to API response model."""
        ...
    
    # CRUD operations
    async def create(self, data: ProjectCreate, author_id: str) -> Project:
        ...
    
    async def get_by_id(self, project_id: str, workspace_id: str) -> Optional[Project]:
        ...
    
    async def update(self, project_id: str, workspace_id: str, data: ProjectUpdate) -> Optional[Project]:
        ...
    
    async def delete(self, project_id: str, workspace_id: str) -> bool:
        ...
    
    async def list_by_workspace(self, workspace_id: str) -> list[Project]:
        ...

# Singleton instance
project_service = ProjectService()
```

---

## Document Conversion Methods

### Document → Model (from Cosmos)

```python
def _doc_to_model_in_db(self, doc: dict) -> ProjectInDB:
    """Convert Cosmos document to ProjectInDB."""
    return ProjectInDB(
        id=doc["id"],
        name=doc["name"],
        description=doc.get("description"),
        slug=doc["slug"],
        workspace_id=doc["workspaceId"],
        author_id=doc["authorId"],
        visibility=doc.get("visibility", "public"),
        tags=doc.get("tags", []),
        created_at=datetime.fromisoformat(doc["createdAt"]),
        updated_at=datetime.fromisoformat(doc["updatedAt"]) if doc.get("updatedAt") else None,
        doc_type=doc.get("docType", "project"),
    )
```

### Model → Document (to Cosmos)

```python
def _model_in_db_to_doc(self, model: ProjectInDB) -> dict:
    """Convert ProjectInDB to Cosmos document."""
    return {
        "id": model.id,
        "name": model.name,
        "description": model.description,
        "slug": model.slug,
        "workspaceId": model.workspace_id,  # Partition key
        "authorId": model.author_id,
        "visibility": model.visibility,
        "tags": model.tags,
        "createdAt": model.created_at.isoformat(),
        "updatedAt": model.updated_at.isoformat() if model.updated_at else None,
        "docType": model.doc_type,
    }
```

### InDB → Response (strip internal fields)

```python
def _model_in_db_to_model(self, model_in_db: ProjectInDB) -> Project:
    """Convert ProjectInDB to API response model."""
    return Project(
        id=model_in_db.id,
        name=model_in_db.name,
        description=model_in_db.description,
        slug=model_in_db.slug,
        workspace_id=model_in_db.workspace_id,
        author_id=model_in_db.author_id,
        visibility=model_in_db.visibility,
        tags=model_in_db.tags,
        created_at=model_in_db.created_at,
        updated_at=model_in_db.updated_at,
    )
```

---

## CRUD Operations

### Create

```python
async def create(self, data: ProjectCreate, author_id: str) -> Project:
    """Create a new project."""
    if not self._use_cosmos():
        raise RuntimeError("Database unavailable")
    
    now = datetime.now(timezone.utc)
    slug = await self._generate_unique_slug(data.name, data.workspace_id)
    
    project_in_db = ProjectInDB(
        id=str(uuid.uuid4()),
        name=data.name,
        description=data.description,
        slug=slug,
        workspace_id=data.workspace_id,
        author_id=author_id,
        visibility=data.visibility,
        tags=data.tags,
        created_at=now,
        updated_at=None,
        doc_type="project",
    )
    
    doc = self._model_in_db_to_doc(project_in_db)
    await upsert_document(doc, partition_key=data.workspace_id)
    
    return self._model_in_db_to_model(project_in_db)
```

### Read (Get by ID)

```python
async def get_by_id(self, project_id: str, workspace_id: str) -> Optional[Project]:
    """Get project by ID. Returns None if not found."""
    if not self._use_cosmos():
        return None
    
    doc = await get_document(project_id, partition_key=workspace_id)
    if doc is None:
        return None
    
    model_in_db = self._doc_to_model_in_db(doc)
    return self._model_in_db_to_model(model_in_db)
```

### Update

```python
async def update(
    self, project_id: str, workspace_id: str, data: ProjectUpdate
) -> Optional[Project]:
    """Update project. Returns None if not found."""
    if not self._use_cosmos():
        return None
    
    doc = await get_document(project_id, partition_key=workspace_id)
    if doc is None:
        return None
    
    model_in_db = self._doc_to_model_in_db(doc)
    
    # Apply updates (only non-None fields)
    update_data = data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        if hasattr(model_in_db, field):
            setattr(model_in_db, field, value)
    
    model_in_db.updated_at = datetime.now(timezone.utc)
    
    updated_doc = self._model_in_db_to_doc(model_in_db)
    await upsert_document(updated_doc, partition_key=workspace_id)
    
    return self._model_in_db_to_model(model_in_db)
```

### Delete

```python
async def delete(self, project_id: str, workspace_id: str) -> bool:
    """Delete project. Returns True if deleted."""
    if not self._use_cosmos():
        return False
    
    return await delete_document(project_id, partition_key=workspace_id)
```

### List

```python
async def list_by_workspace(self, workspace_id: str) -> list[Project]:
    """List all projects in a workspace."""
    if not self._use_cosmos():
        return []
    
    docs = await query_documents(
        doc_type="project",
        partition_key=workspace_id,
    )
    
    return [
        self._model_in_db_to_model(self._doc_to_model_in_db(doc))
        for doc in docs
    ]
```

---

## Query Patterns

### Basic Query with Filter

```python
async def get_by_slug(self, slug: str, workspace_id: str) -> Optional[Project]:
    """Get project by slug within a workspace."""
    docs = await query_documents(
        doc_type="project",
        partition_key=workspace_id,
        extra_filter="AND c.slug = @slug",
        parameters=[{"name": "@slug", "value": slug}],
    )
    
    if not docs:
        return None
    
    return self._model_in_db_to_model(self._doc_to_model_in_db(docs[0]))
```

### Unique Slug Generation

```python
async def _generate_unique_slug(self, name: str, workspace_id: str) -> str:
    """Generate unique slug from name."""
    base_slug = slugify(name)
    slug = base_slug
    counter = 1
    
    while True:
        existing = await query_documents(
            doc_type="project",
            partition_key=workspace_id,
            extra_filter="AND c.slug = @slug",
            parameters=[{"name": "@slug", "value": slug}],
        )
        if not existing:
            return slug
        slug = f"{base_slug}-{counter}"
        counter += 1
```

---

## Graceful Degradation

Every public method checks `_use_cosmos()` and returns safe defaults:

| Return Type | Default |
|-------------|---------|
| `Optional[Model]` | `None` |
| `list[Model]` | `[]` |
| `bool` | `False` |
| Required create | `raise RuntimeError` |

```python
async def get_by_id(self, project_id: str, workspace_id: str) -> Optional[Project]:
    if not self._use_cosmos():
        return None  # Graceful None instead of exception
    ...
```

---

## Complete Example

See [assets/service_template.py](../assets/service_template.py) for a production-ready service implementation.

````


### `references/testing.md`

````markdown
# Testing Cosmos DB Services

## Table of Contents

1. [Test-Driven Development Workflow](#test-driven-development-workflow)
2. [Test File Structure](#test-file-structure)
3. [Fixtures and Mocking](#fixtures-and-mocking)
4. [Unit Test Patterns](#unit-test-patterns)
5. [Integration Test Patterns](#integration-test-patterns)
6. [Test Data Factories](#test-data-factories)

---

## Test-Driven Development Workflow

Follow Red-Green-Refactor:

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

### Example TDD Cycle

```python
# 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

```python
"""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)

```python
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

```python
@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

```python
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

```python
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

```python
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

```python
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

```python
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):

```python
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:

```python
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](../assets/conftest_template.py) for a production-ready conftest.py.

````