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

1import dataclasses 

2from typing import ( 

3 TypeVar, 

4 TYPE_CHECKING, 

5 Tuple, 

6 Dict, 

7 Optional, 

8 TypedDict, 

9 NotRequired, 

10 List, 

11) 

12from collections.abc import Sequence, Mapping, MutableMapping 

13 

14if TYPE_CHECKING: 

15 from debputy.plugin.api import VirtualPath 

16 from debputy.filesystem_scan import FSPath 

17 

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

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

20else: 

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

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

23 

24 

25class EnvironmentModificationSerialized(TypedDict): 

26 replacements: NotRequired[dict[str, str]] 

27 removals: NotRequired[list[str]] 

28 

29 

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

31class EnvironmentModification: 

32 replacements: Sequence[tuple[str, str]] = tuple() 

33 removals: Sequence[str] = tuple() 

34 

35 def __bool__(self) -> bool: 

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

37 

38 def combine( 

39 self, 

40 other: "Optional[EnvironmentModification]", 

41 ) -> "EnvironmentModification": 

42 if not other: 

43 return self 

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

45 extra_replacements = { 

46 k: v 

47 for k, v in other.replacements 

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

49 } 

50 seen_removals = set(self.removals) 

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

52 

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

54 new_replacements = self.replacements 

55 else: 

56 combined_replacements = [] 

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

58 if k not in extra_replacements: 

59 combined_replacements.append((k, v)) 

60 

61 for k, v in other.replacements: 

62 if k in extra_replacements: 

63 combined_replacements.append((k, v)) 

64 

65 new_replacements = tuple(combined_replacements) 

66 

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

68 new_removals = self.removals 

69 else: 

70 combined_removals = list(self.removals) 

71 combined_removals.extend(extra_removals) 

72 new_removals = tuple(combined_removals) 

73 

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

75 return self 

76 

77 return EnvironmentModification( 

78 new_replacements, 

79 new_removals, 

80 ) 

81 

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

83 for k, v in self.replacements: 

84 existing_value = env.get(k) 

85 if v == existing_value: 

86 continue 

87 env[k] = v 

88 

89 for k in self.removals: 

90 if k not in env: 

91 continue 

92 del env[k] 

93 

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

95 updated_env: dict[str, str] | None = None 

96 for k, v in self.replacements: 

97 existing_value = base_env.get(k) 

98 if v == existing_value: 

99 continue 

100 

101 if updated_env is None: 

102 updated_env = dict(base_env) 

103 updated_env[k] = v 

104 

105 for k in self.removals: 

106 if k not in base_env: 

107 continue 

108 if updated_env is None: 

109 updated_env = dict(base_env) 

110 del updated_env[k] 

111 

112 if updated_env is not None: 

113 return updated_env 

114 return base_env