Handling OTA API Rate Limits in Rate Parity Automation

import asyncio
import logging
import time
from datetime import datetime, timezone
from typing import Dict, Any, Optional

import aiohttp
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_log,
    after_log
)

# Structured logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='{"time":"%(asctime)s","level":"%(levelname)s","logger":"%(name)s","message":"%(message)s","otel.trace_id":"%(otelTraceId)s"}'
)
logger = logging.getLogger("ota_rate_parity")

class OTARateDispatcher:
    def __init__(self, base_url: str, api_key: str, max_concurrency: int = 5):
        self.base_url = base_url
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.retry_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()

    def _build_headers(self, token: str) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "X-Property-ID": "PROP_8842"
        }

    @retry(
        retry=retry_if_exception_type((aiohttp.ClientResponseError, asyncio.TimeoutError)),
        wait=wait_exponential(multiplier=1, min=2, max=60),
        stop=stop_after_attempt(5),
        before=before_log(logger, logging.DEBUG),
        after=after_log(logger, logging.INFO)
    )
    async def _dispatch_rate_update(self, session: aiohttp.ClientSession, payload: Dict[str, Any]) -> bool:
        async with self.semaphore:
            try:
                async with session.post(
                    f"{self.base_url}/rates/update",
                    json=payload,
                    headers=self._build_headers("cached_token_placeholder"),
                    timeout=aiohttp.ClientTimeout(total=10)
                ) as response:
                    if response.status == 429:
                        retry_after = int(response.headers.get("Retry-After", 10))
                        logger.warning(
                            "Rate limit hit. Backing off.",
                            extra={"retry_after_seconds": retry_after, "status": 429}
                        )
                        raise aiohttp.ClientResponseError(
                            request_info=response.request_info,
                            history=response.history,
                            status=429,
                            message="Too Many Requests"
                        )
                    elif response.status == 401:
                        logger.error("Token expired. Triggering silent refresh.")
                        raise aiohttp.ClientResponseError(
                            request_info=response.request_info,
                            history=response.history,
                            status=401,
                            message="Unauthorized"
                        )
                    elif response.status >= 400:
                        logger.error("OTA API error", extra={"status": response.status, "payload_id": payload.get("id")})
                        return False

                    logger.info("Rate parity update successful", extra={"payload_id": payload.get("id")})
                    return True
            except asyncio.TimeoutError:
                logger.warning("Request timed out. Retrying...")
                raise

    async def enqueue_update(self, priority: int, payload: Dict[str, Any]):
        """Pushes a validated rate payload into the async dispatch queue."""
        await self.retry_queue.put((priority, payload))
        logger.info("Payload enqueued", extra={"priority": priority, "rate_plan": payload.get("rate_plan_id")})

    async def drain_queue(self):
        """Processes queued updates with concurrency control and backoff."""
        while not self.retry_queue.empty():
            priority, payload = await self.retry_queue.get()
            try:
                async with aiohttp.ClientSession() as session:
                    await self._dispatch_rate_update(session, payload)
            except Exception as e:
                logger.error("Final dispatch failure. Persisting to dead-letter queue.", extra={"error": str(e)})
            finally:
                self.retry_queue.task_done()

# Usage Context
async def main():
    dispatcher = OTARateDispatcher("https://api.channelmanager.example/v2", "sk_live_...")

    # Pre-flight validation would occur here (delta checks, blackout alignment, inventory bounds)
    sample_payload = {"rate_plan_id": "RP_101", "date": "2024-12-25", "price": 189.50, "inventory": 12}
    await dispatcher.enqueue_update(priority=1, payload=sample_payload)

    await dispatcher.drain_queue()

if __name__ == "__main__":
    asyncio.run(main())