Coverage for src/debputy/types.py: 16%

1import dataclasses 

2from typing import ( 

3 TypeVar, 

4 TYPE_CHECKING, 

5 Sequence, 

6 Tuple, 

7 Mapping, 

8 Dict, 

9 Optional, 

10 TypedDict, 

11 NotRequired, 

12 List, 

13 MutableMapping, 

14) 

15 

16if TYPE_CHECKING: 

17 from debputy.plugin.api import VirtualPath 

18 from debputy.filesystem_scan import FSPath 

19 

20 VP = TypeVar("VP", VirtualPath, FSPath) 

21 S = TypeVar("S", str, bytes) 

22else: 

23 VP = TypeVar("VP", "VirtualPath", "FSPath") 

24 S = TypeVar("S", str, bytes) 

25 

26 

27class EnvironmentModificationSerialized(TypedDict): 

28 replacements: NotRequired[Dict[str, str]] 

29 removals: NotRequired[List[str]] 

30 

31 

32@dataclasses.dataclass(slots=True, frozen=True) 

33class EnvironmentModification: 

34 replacements: Sequence[Tuple[str, str]] = tuple() 

35 removals: Sequence[str] = tuple() 

36 

37 def __bool__(self) -> bool: 

38 return bool(self.removals) or bool(self.replacements) 

39 

40 def combine( 

41 self, 

42 other: "Optional[EnvironmentModification]", 

43 ) -> "EnvironmentModification": 

44 if not other: 

45 return self 

46 existing_replacements = {k: v for k, v in self.replacements} 

47 extra_replacements = { 

48 k: v 

49 for k, v in other.replacements 

50 if k not in existing_replacements or existing_replacements[k] != v 

51 } 

52 seen_removals = set(self.removals) 

53 extra_removals = [r for r in other.removals if r not in seen_removals] 

54 

55 if not extra_replacements and isinstance(self.replacements, tuple): 

56 new_replacements = self.replacements 

57 else: 

58 combined_replacements = [] 

59 for k, v in existing_replacements.items(): 

60 if k not in extra_replacements: 

61 combined_replacements.append((k, v)) 

62 

63 for k, v in other.replacements: 

64 if k in extra_replacements: 

65 combined_replacements.append((k, v)) 

66 

67 new_replacements = tuple(combined_replacements) 

68 

69 if not extra_removals and isinstance(self.removals, tuple): 

70 new_removals = self.removals 

71 else: 

72 combined_removals = list(self.removals) 

73 combined_removals.extend(extra_removals) 

74 new_removals = tuple(combined_removals) 

75 

76 if self.replacements is new_replacements and self.removals is new_removals: 

77 return self 

78 

79 return EnvironmentModification( 

80 new_replacements, 

81 new_removals, 

82 ) 

83 

84 def update_inplace(self, env: MutableMapping[str, str]) -> None: 

85 for k, v in self.replacements: 

86 existing_value = env.get(k) 

87 if v == existing_value: 

88 continue 

89 env[k] = v 

90 

91 for k in self.removals: 

92 if k not in env: 

93 continue 

94 del env[k] 

95 

96 def compute_env(self, base_env: Mapping[str, str]) -> Mapping[str, str]: 

97 updated_env: Optional[Dict[str, str]] = None 

98 for k, v in self.replacements: 

99 existing_value = base_env.get(k) 

100 if v == existing_value: 

101 continue 

102 

103 if updated_env is None: 

104 updated_env = dict(base_env) 

105 updated_env[k] = v 

106 

107 for k in self.removals: 

108 if k not in base_env: 

109 continue 

110 if updated_env is None: 

111 updated_env = dict(base_env) 

112 del updated_env[k] 

113 

114 if updated_env is not None: 

115 return updated_env 

116 return base_env