from typing import Optional, Union
from multidict import MultiDictProxy
from harp.typing.storage import Storage
from harp_apps.dashboard.filters.utils import flatten_facet_value, str_to_float_or_none
[docs]
class AbstractFacet:
name = "name"
[docs]
def __init__(self):
self.meta = {}
@property
def values(self):
raise NotImplementedError
[docs]
def get_filter(self, raw_data: list):
raise NotImplementedError
[docs]
def filter(self, raw_data: list):
raise NotImplementedError
[docs]
def get_filter_from_query(self, query: MultiDictProxy):
raise NotImplementedError
[docs]
def filter_from_query(self, query: MultiDictProxy):
raise NotImplementedError
[docs]
class AbstractChoicesFacet(AbstractFacet):
choices: Union[set | list] = set()
exhaustive = True
@property
def values(self):
return [{"name": choice, "count": self.meta.get(choice, {}).get("count", None)} for choice in self.choices]
[docs]
def get_filter(self, raw_data: list):
query_endpoints = set(self.choices).intersection(raw_data)
return list(query_endpoints) if len(query_endpoints) else None
[docs]
def get_filter_from_query(self, query: MultiDictProxy):
raw_data = self._choices_from_query(query)
return self.get_filter(raw_data)
[docs]
def filter(self, raw_data: list):
return {
"values": self.values,
"current": self.get_filter(raw_data),
}
[docs]
def filter_from_query(self, query: MultiDictProxy):
raw_data = self._choices_from_query(query)
return self.filter(raw_data)
def _choices_from_query(self, query: MultiDictProxy):
return flatten_facet_value(query.getall(self.name, []))
[docs]
class FacetWithStorage(AbstractChoicesFacet):
[docs]
def __init__(self, *, storage: Storage):
super().__init__()
self.storage = storage
[docs]
class AbstractMinMaxFacet(AbstractFacet):
min: float = 0.0
max: float = 100.0
[docs]
def __init__(self):
self.meta = {}
@property
def values(self):
return {
"min": self.meta.get("min", None),
"max": self.meta.get("max", None),
}
[docs]
def filter(self, min, max):
return {
"values": ["min", "max"],
"current": self.get_filter(min, max),
}
[docs]
def filter_from_query(self, query: MultiDictProxy):
min, max = self._min_max_from_query(query)
return self.filter(min, max)
[docs]
def get_filter(self, min: Optional[float], max: Optional[float]):
return {"min": min, "max": max}
[docs]
def get_filter_from_query(self, query: MultiDictProxy):
min, max = self._min_max_from_query(query)
return self.get_filter(min, max)
def _min_max_from_query(self, query: MultiDictProxy):
return str_to_float_or_none(query.get(self.name + "min", "")), str_to_float_or_none(
query.get(self.name + "max", "")
)
[docs]
class NonExhaustiveFacet(AbstractChoicesFacet):
exhaustive = False
fallback_name = "NULL"
[docs]
def __init__(self):
super().__init__()
self.choices = list(self.choices) + [
"NULL",
]
[docs]
def filter(self, raw_data: list):
return {
"values": self.values,
"current": self.get_filter(raw_data),
"fallbackName": self.fallback_name,
}