Coverage for ramose / filters.py: 100%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7from collections.abc import Mapping 

8from pathlib import Path 

9from re import sub 

10 

11import yaml 

12 

13FiltersConfig = Mapping[str, Mapping[str, "str | dict[str, str]"]] 

14 

15_VALUE_PATTERN = r"\{\{\s*value\s*\}\}" 

16_ALWAYS_EMPTY_FILTER = "FILTER(false)" 

17 

18 

19def render(template: str, value: str) -> str: 

20 return sub(_VALUE_PATTERN, value, template) 

21 

22 

23def _select_template(key: str, spec: str | dict[str, str], value: str) -> str: 

24 if isinstance(spec, str): 

25 return spec 

26 if value not in spec: 

27 msg = f"The value '{value}' is not valid for filter '{key}', valid values are {', '.join(sorted(spec))}" 

28 raise ValueError(msg) 

29 return spec[value] 

30 

31 

32def _is_always_empty(fragment: str) -> bool: 

33 return fragment.strip().upper() == _ALWAYS_EMPTY_FILTER.upper() 

34 

35 

36def apply_filters(config: FiltersConfig, values: list[str]) -> dict[str, str]: 

37 slots: list[str] = [] 

38 for slot_map in config.values(): 

39 for slot in slot_map: 

40 if slot not in slots: 

41 slots.append(slot) 

42 result = dict.fromkeys(slots, "") 

43 fragments: list[tuple[str, str]] = [] 

44 

45 raw_pairs = (pair.strip() for value in values for pair in value.split(",")) 

46 for pair in (pair for pair in raw_pairs if pair): 

47 key, value = pair.split(":", 1) 

48 if key not in config: 

49 msg = f"The filter '{key}' is not configured, configured filters are {', '.join(sorted(config))}" 

50 raise ValueError(msg) 

51 for slot, spec in config[key].items(): 

52 fragment = render(_select_template(key, spec, value), value) 

53 fragments.append((slot, fragment)) 

54 

55 always_empty_slots = {slot for slot, fragment in fragments if _is_always_empty(fragment)} 

56 if always_empty_slots: 

57 for slot in always_empty_slots: 

58 result[slot] = _ALWAYS_EMPTY_FILTER 

59 return result 

60 

61 for slot, fragment in fragments: 

62 result[slot] = f"{result[slot]}\n{fragment}" if result[slot] else fragment 

63 return result 

64 

65 

66def load_filters_config(path: str) -> FiltersConfig: 

67 return yaml.safe_load(Path(path).read_text(encoding="utf-8"))