utils/api_utils.py

"""
API utility functions and classes for the TellusDigital API Response Comparator
"""

import json
from typing import Any

from utils.logger import structlog

log = structlog.get_logger()

class CustomJSONEncoder(json.JSONEncoder):
    """Custom JSON encoder that handles DeepDiff types."""
    
    def default(self, obj: Any) -> Any:
        """Convert DeepDiff types to JSON-serializable format."""
        try:
            values = getattr(obj, '_values', None)
            if values is not None:
                return list(values)
            return json.JSONEncoder.default(self, obj)
        except TypeError:
            return str(obj)

async def broadcast_progress(active_connections: list, message: str) -> None:
    """Broadcast progress updates to all connected WebSocket clients."""
    connections = active_connections.copy()
    for connection in connections:
        try:
            await connection.send_json({"type": "processing", "status": message})
        except Exception as e:
            log.error("Broadcast error", error=str(e))
            if connection in active_connections:
                active_connections.remove(connection)