from typing import FrozenSet, List, Tuple, Union from cachetools import cached from delphi_api.core.caches import CompositeRangeCache, ItemKeysCache class QueryUtils: """ Legacy static data and query utility functions. Used by v2 and unit tests. This class is deprecated and should not be added to. """ @staticmethod @cached(ItemKeysCache) def build_item_keys(hash_keys: Union[FrozenSet[str], FrozenSet[List[str]]], range_keys: Union[FrozenSet[str], FrozenSet[List[str]]] = None, other_keys: Union[FrozenSet[str], FrozenSet[List[str]]] = None, ) -> Union[List[Tuple[str, str]], List[Tuple[str]]]: """Builds tuples of hash keys and range keys for batch get operations. Args: hash_keys: set of hash (aka partition) keys range_keys: (optional) set of range (aka sort) keys other_keys: (optional) set of other keys Returns: List[Tuple[str, str]]: a list of tuples as ``[(hash_key, range_key), ...]`` """ item_keys: List[tuple] = [] hash_keys = sorted(hash_keys) range_keys = sorted(range_keys) if range_keys else None other_keys = sorted(other_keys) if other_keys else None if not range_keys: return [(hash_key,) for hash_key in hash_keys] for range_key in range_keys: for hash_key in hash_keys: if not other_keys: item_keys.append((hash_key, range_key)) continue for other_key in other_keys: item_keys.append((hash_key, range_key, other_key)) return item_keys @staticmethod @cached(CompositeRangeCache) def build_composite_range_keys(set1: FrozenSet[str], set2: FrozenSet[str], set3: FrozenSet[str] = None) -> List[str]: """Composes range keys via enumeration and concatenation. Potentially slow depending on input. Examples: .. code-block:: python dates: Set[str] = {...} country_codes: Set[str] = {...} container_id: Set[str] = {...} result = QueryUtils.build_composite_range_keys(dates, country_codes, container_id) # builds { {Date}_{CountryCode}_{ContainerId} } Returns: List[str]: a list of permutations [ {arg[0]}_{...}_{arg[n]} ] """ range_keys = [] set1 = sorted(set1) set2 = sorted(set2) set3 = sorted(set3) if set3 else None for item1 in set1: for item2 in set2: tmp_str = f'{item1}_{item2}' if not set3: range_keys.append(tmp_str) continue for item3 in set3: range_keys.append(f'{tmp_str}_{item3}') return range_keys