services.task_dispatcher

 1from typing import Any, Dict, List, Tuple
 2
 3from services.api_clients import AgifyClient, JokeClient, PostmanClient
 4from utils.logger import info_logger, error_logger
 5
 6
 7class AsyncTaskDispatcher:
 8    """Dispatches tasks to external API clients and manages caching for age predictions."""
 9
10    def __init__(self) -> None:
11        """Initialize API clients and internal age cache."""
12        self.age_cache: Dict[Tuple[str, str], Dict[str, Any]] = {}
13        self.agify_client = AgifyClient()
14        self.joke_client = JokeClient()
15        self.postman_client = PostmanClient()
16
17    async def preload_age_predictions(
18        self, name_country_pairs: List[Tuple[str, str]]
19    ) -> None:
20        """
21        Preload and cache age predictions for batches of (name, country) pairs.
22
23        :param name_country_pairs: List of (name, country) tuples to preload predictions for.
24        """
25        country_groups: Dict[str, List[str]] = {}
26        for name, country in name_country_pairs:
27            country = country.upper()
28            country_groups.setdefault(country, []).append(name)
29
30        for country, names in country_groups.items():
31            for i in range(0, len(names), 10):
32                batch = names[i : i + 10]
33                try:
34                    results = await self.agify_client.get_batch_ages(batch, country)
35                    info_logger.info(
36                        f"[BATCH] {len(batch)} names in {country}: {', '.join(batch)}"
37                    )
38                    for result in results:
39                        self.age_cache[(result["name"], country)] = result
40                except Exception as e:
41                    error_logger.error(
42                        f"[BATCH] Failed batch request for {country}: {str(e)}"
43                    )
44
45    async def handle(self, data: Dict[str, Any]) -> Dict[str, Any]:
46        """
47        Process a single task based on its type ("age", "joke", etc.).
48
49        :param data: Input data dict with a "type" field.
50        :return: Processed result posted through PostmanClient.
51        :raises Exception: If any error occurs during processing.
52        """
53        task_type: str = data.get("type", "").lower()
54        name: str = data.get("name", "")
55        country: str = data.get("country", "").upper()
56
57        try:
58            if task_type == "age":
59                key = (name, country)
60                if key not in self.age_cache:
61                    result = await self.agify_client.get_age(name, country)
62                    info_logger.info(f"[SINGLE] Age fetched for {name} in {country}")
63                    self.age_cache[key] = result
64                response: Dict[str, Any] = self.age_cache[key]
65
66            elif task_type == "joke":
67                response = await self.joke_client.get_random_joke()
68                info_logger.info(f"[JOKE] Random joke fetched")
69
70            else:
71                response = data
72                info_logger.info(
73                    f"[RAW] Unrecognized task_type. Used input as response."
74                )
75
76            postman_response = await self.postman_client.post_response(response)
77            return postman_response.get("json", {})
78
79        except Exception as e:
80            error_logger.error(f"[TASK] Failed to handle task: {str(e)}")
81            raise