|
|
| from typing import Any, Dict, Optional, Tuple
|
|
|
|
|
| class EntrySelector:
|
| """
|
| Base class for entry selectors
|
| """
|
|
|
| @staticmethod
|
| def from_string(spec: str) -> "EntrySelector":
|
| if spec == "*":
|
| return AllEntrySelector()
|
| return FieldEntrySelector(spec)
|
|
|
|
|
| class AllEntrySelector(EntrySelector):
|
| """
|
| Selector that accepts all entries
|
| """
|
|
|
| SPECIFIER = "*"
|
|
|
| def __call__(self, entry):
|
| return True
|
|
|
|
|
| class FieldEntrySelector(EntrySelector):
|
| """
|
| Selector that accepts only entries that match provided field
|
| specifier(s). Only a limited set of specifiers is supported for now:
|
| <specifiers>::=<specifier>[<comma><specifiers>]
|
| <specifier>::=<field_name>[<type_delim><type>]<equal><value_or_range>
|
| <field_name> is a valid identifier
|
| <type> ::= "int" | "str"
|
| <equal> ::= "="
|
| <comma> ::= ","
|
| <type_delim> ::= ":"
|
| <value_or_range> ::= <value> | <range>
|
| <range> ::= <value><range_delim><value>
|
| <range_delim> ::= "-"
|
| <value> is a string without spaces and special symbols
|
| (e.g. <comma>, <equal>, <type_delim>, <range_delim>)
|
| """
|
|
|
| _SPEC_DELIM = ","
|
| _TYPE_DELIM = ":"
|
| _RANGE_DELIM = "-"
|
| _EQUAL = "="
|
| _ERROR_PREFIX = "Invalid field selector specifier"
|
|
|
| class _FieldEntryValuePredicate:
|
| """
|
| Predicate that checks strict equality for the specified entry field
|
| """
|
|
|
| def __init__(self, name: str, typespec: Optional[str], value: str):
|
| import builtins
|
|
|
| self.name = name
|
| self.type = getattr(builtins, typespec) if typespec is not None else str
|
| self.value = value
|
|
|
| def __call__(self, entry):
|
| return entry[self.name] == self.type(self.value)
|
|
|
| class _FieldEntryRangePredicate:
|
| """
|
| Predicate that checks whether an entry field falls into the specified range
|
| """
|
|
|
| def __init__(self, name: str, typespec: Optional[str], vmin: str, vmax: str):
|
| import builtins
|
|
|
| self.name = name
|
| self.type = getattr(builtins, typespec) if typespec is not None else str
|
| self.vmin = vmin
|
| self.vmax = vmax
|
|
|
| def __call__(self, entry):
|
| return (entry[self.name] >= self.type(self.vmin)) and (
|
| entry[self.name] <= self.type(self.vmax)
|
| )
|
|
|
| def __init__(self, spec: str):
|
| self._predicates = self._parse_specifier_into_predicates(spec)
|
|
|
| def __call__(self, entry: Dict[str, Any]):
|
| for predicate in self._predicates:
|
| if not predicate(entry):
|
| return False
|
| return True
|
|
|
| def _parse_specifier_into_predicates(self, spec: str):
|
| predicates = []
|
| specs = spec.split(self._SPEC_DELIM)
|
| for subspec in specs:
|
| eq_idx = subspec.find(self._EQUAL)
|
| if eq_idx > 0:
|
| field_name_with_type = subspec[:eq_idx]
|
| field_name, field_type = self._parse_field_name_type(field_name_with_type)
|
| field_value_or_range = subspec[eq_idx + 1 :]
|
| if self._is_range_spec(field_value_or_range):
|
| vmin, vmax = self._get_range_spec(field_value_or_range)
|
| predicate = FieldEntrySelector._FieldEntryRangePredicate(
|
| field_name, field_type, vmin, vmax
|
| )
|
| else:
|
| predicate = FieldEntrySelector._FieldEntryValuePredicate(
|
| field_name, field_type, field_value_or_range
|
| )
|
| predicates.append(predicate)
|
| elif eq_idx == 0:
|
| self._parse_error(f'"{subspec}", field name is empty!')
|
| else:
|
| self._parse_error(f'"{subspec}", should have format ' "<field>=<value_or_range>!")
|
| return predicates
|
|
|
| def _parse_field_name_type(self, field_name_with_type: str) -> Tuple[str, Optional[str]]:
|
| type_delim_idx = field_name_with_type.find(self._TYPE_DELIM)
|
| if type_delim_idx > 0:
|
| field_name = field_name_with_type[:type_delim_idx]
|
| field_type = field_name_with_type[type_delim_idx + 1 :]
|
| elif type_delim_idx == 0:
|
| self._parse_error(f'"{field_name_with_type}", field name is empty!')
|
| else:
|
| field_name = field_name_with_type
|
| field_type = None
|
|
|
|
|
| return field_name, field_type
|
|
|
| def _is_range_spec(self, field_value_or_range):
|
| delim_idx = field_value_or_range.find(self._RANGE_DELIM)
|
| return delim_idx > 0
|
|
|
| def _get_range_spec(self, field_value_or_range):
|
| if self._is_range_spec(field_value_or_range):
|
| delim_idx = field_value_or_range.find(self._RANGE_DELIM)
|
| vmin = field_value_or_range[:delim_idx]
|
| vmax = field_value_or_range[delim_idx + 1 :]
|
| return vmin, vmax
|
| else:
|
| self._parse_error('"field_value_or_range", range of values expected!')
|
|
|
| def _parse_error(self, msg):
|
| raise ValueError(f"{self._ERROR_PREFIX}: {msg}")
|
|
|
