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

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

"""
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()

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

# 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,
)
```

# 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"}],
)
```

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

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