Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/events/schemas/labelling.py: 36%

58 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 11:21 +0000

1from typing import Dict, Iterable, Iterator, List, Optional, Tuple 1a

2 

3from pydantic import RootModel 1a

4 

5 

6class LabelDiver: 1a

7 """The LabelDiver supports templating use cases for any Labelled object, by 

8 presenting the labels as a graph of objects that may be accessed by attribute. For 

9 example: 

10 

11 ```python 

12 diver = LabelDiver({ 

13 'hello.world': 'foo', 

14 'hello.world.again': 'bar' 

15 }) 

16 

17 assert str(diver.hello.world) == 'foo' 

18 assert str(diver.hello.world.again) == 'bar' 

19 ``` 

20 

21 """ 

22 

23 _value: str 1a

24 _divers: Dict[str, "LabelDiver"] 1a

25 _labels: Dict[str, str] 1a

26 

27 def __init__(self, labels: Dict[str, str], value: str = ""): 1a

28 self._labels = labels.copy() 

29 self._value = value 

30 

31 divers: Dict[str, Dict[str, str]] = {} 

32 values: Dict[str, str] = {} 

33 

34 for key, value in labels.items(): 

35 head, _, tail = key.partition(".") 

36 if tail: 

37 if head not in divers: 

38 divers[head] = {} 

39 

40 divers[head][tail] = labels[key] 

41 else: 

42 values[head] = value 

43 

44 # start with keys that had sub-divers... 

45 self._divers: Dict[str, LabelDiver] = { 

46 k: LabelDiver(v, value=values.pop(k, "")) for k, v in divers.items() 

47 } 

48 # ...then mix in any remaining keys that _only_ had values 

49 self._divers.update(**{k: LabelDiver({}, value=v) for k, v in values.items()}) 

50 

51 def __str__(self) -> str: 1a

52 return self._value or "" 

53 

54 def __repr__(self) -> str: 1a

55 return f"LabelDiver(divers={self._divers!r}, value={self._value!r})" 

56 

57 def __len__(self) -> int: 1a

58 return len(self._labels) 

59 

60 def __iter__(self) -> Iterator[Tuple[str, str]]: 1a

61 return iter(self._labels.items()) 

62 

63 def __getitem__(self, key: str) -> str: 1a

64 return self._labels[key] 

65 

66 def __getattr__(self, name: str) -> "LabelDiver": 1a

67 if name.startswith("_"): 

68 raise AttributeError 

69 

70 try: 

71 return self._divers[name] 

72 except KeyError: 

73 raise AttributeError 

74 

75 

76class Labelled(RootModel[Dict[str, str]]): 1a

77 def keys(self) -> Iterable[str]: 1a

78 return self.root.keys() 

79 

80 def items(self) -> Iterable[Tuple[str, str]]: 1a

81 return self.root.items() 

82 

83 def __getitem__(self, label: str) -> str: 1a

84 return self.root[label] 

85 

86 def __setitem__(self, label: str, value: str) -> str: 1a

87 self.root[label] = value 

88 return value 

89 

90 def __contains__(self, key: str) -> bool: 1a

91 return key in self.root 

92 

93 def get(self, label: str, default: Optional[str] = None) -> Optional[str]: 1a

94 return self.root.get(label, default) 

95 

96 def as_label_value_array(self) -> List[Dict[str, str]]: 1a

97 return [{"label": label, "value": value} for label, value in self.items()] 

98 

99 @property 1a

100 def labels(self) -> LabelDiver: 1a

101 return LabelDiver(self.root) 

102 

103 def has_all_labels(self, labels: Dict[str, str]) -> bool: 1a

104 return all(self.root.get(label) == value for label, value in labels.items())