import itertools from datetime import date from typing import Mapping, List, Dict, Any, Iterable, Optional, Tuple __all__ = ["ParamsParser"] class ParamsParser: prefix_map: Mapping[str, str] = {"min": "gte", "max": "lte"} @classmethod def parse(cls, params: Dict[str, Any]) -> Iterable[Dict[str, Any]]: return cls._parse(params) @staticmethod def _group_key_func(key: str) -> str: if key.startswith("min_") or key.startswith("max_"): return key[4:] return key @staticmethod def _build_param_key(*parts: str) -> str: return ".".join(parts) @classmethod def _handle_single_param(cls, *parents: str, key: str, value: Any) -> Optional[Dict[str, Any]]: pair = cls._transform_key_value(key, value) if pair is None and parents: return {"exists": {"field": parents[0]}} if pair is None: return None key, value = pair return {"match": {cls._build_param_key(*parents, key): value}} @classmethod def _handle_list_param(cls, key: str, values: List[str]) -> Optional[Dict]: res: Dict = {"bool": {"should": []}} pair = cls._transform_key_value(key, values) if pair is None: return None key, values = pair res["bool"]["must"] = {"bool": {"should": [{"terms": {key: values}}, {"terms": {f"{key}.keyword": values}}]}} return res @classmethod def _handle_group_param(cls, *parents: str, items: Iterable[Tuple[str, Any]]) -> Optional[Dict[str, Any]]: range_items = {} res_key = None for k, v in items: pair = cls._transform_key_value(k, v) if pair is None: continue k, v = pair prefix, key = k.split("_", 1) if res_key is None: res_key = key elif key != res_key: raise ValueError(f"Can't get common key part of '{res_key}' and '{key}'") range_items[cls.prefix_map[prefix]] = v if not range_items or not res_key: return None return {"range": {cls._build_param_key(*parents, res_key): range_items}} @classmethod def _transform_key_value(cls, key: str, value: Any) -> Optional[Tuple[str, Any]]: """ Can be overriden to change key and/or value somehow. returning None means skip this parameter :param key: :param value: :return: """ if key == "chart_type": return None if key == "min_latest_release_days": if value is None: return None key = "min_latest_release_date" value = date.today() - value if key == "countries": key = "country" return key, value @classmethod def _parse(cls, params: Dict[str, Any], *parents: str): ordered = sorted(params.keys(), key=cls._group_key_func) for key, grp in itertools.groupby(ordered, key=cls._group_key_func): # single param if key in params: value = params[key] if isinstance(value, dict): yield from cls._parse(value, *parents, key) elif isinstance(value, list): param = cls._handle_list_param(key, value) if param: yield param else: param = cls._handle_single_param(key=key, value=value, *parents) if param: yield param # range param else: param = cls._handle_group_param(items=((k, params[k]) for k in grp), *parents) if param: yield param