Coverage for src/objectstate/object_state_metadata.py: 84%

1"""Contracted metadata storage for ObjectState extensions. 

2 

3ObjectState.metadata is intentionally outside normal dirty detection, but it is 

4inside time-travel snapshots. Extension-owned metadata therefore needs a 

5contract: namespaced keys must be registered before they can be written, 

6captured, or restored. 

7""" 

8 

9from __future__ import annotations 

10 

11from dataclasses import dataclass 

12from typing import Any, ClassVar, Mapping 

13import copy 

14 

15 

16@dataclass(frozen=True) 

17class ObjectStateMetadataContract: 

18 """Registered extension-owned ObjectState metadata key.""" 

19 

20 key: str 

21 owner: str 

22 description: str 

23 

24 def __post_init__(self) -> None: 

25 if "." not in self.key: 

26 raise ValueError( 

27 f"ObjectState metadata extension key must be namespaced: {self.key!r}" 

28 ) 

29 if not self.owner: 

30 raise ValueError(f"ObjectState metadata contract {self.key!r} needs an owner") 

31 if not self.description: 

32 raise ValueError( 

33 f"ObjectState metadata contract {self.key!r} needs a description" 

34 ) 

35 

36 

37class ObjectStateMetadataContractRegistry: 

38 """Process-local registry for extension-owned ObjectState metadata.""" 

39 

40 _contracts: ClassVar[dict[str, ObjectStateMetadataContract]] = {} 

41 

42 @classmethod 

43 def register(cls, contract: ObjectStateMetadataContract) -> ObjectStateMetadataContract: 

44 existing = cls._contracts.get(contract.key) 

45 if existing is not None and existing != contract: 

46 raise ValueError( 

47 f"ObjectState metadata key {contract.key!r} is already registered " 

48 f"to {existing.owner!r}" 

49 ) 

50 cls._contracts[contract.key] = contract 

51 return contract 

52 

53 @classmethod 

54 def contract_for(cls, key: str) -> ObjectStateMetadataContract: 

55 contract = cls._contracts.get(key) 

56 if contract is None: 

57 raise KeyError( 

58 f"ObjectState extension metadata key {key!r} is not registered. " 

59 "Register ObjectStateMetadataContract before writing namespaced metadata." 

60 ) 

61 return contract 

62 

63 @classmethod 

64 def validate_key(cls, key: str) -> None: 

65 if "." in key: 

66 cls.contract_for(key) 

67 

68 @classmethod 

69 def validate_mapping( 

70 cls, 

71 *, 

72 scope_id: str | None, 

73 metadata: Mapping[str, Any], 

74 operation: str, 

75 ) -> None: 

76 for key in metadata: 

77 try: 

78 cls.validate_key(key) 

79 except KeyError as exc: 

80 scope = scope_id if scope_id is not None else "<root>" 

81 raise RuntimeError( 

82 f"Cannot {operation} ObjectState metadata for {scope}: {exc}" 

83 ) from exc 

84 

85 

86class ObjectStateMetadataStore(dict): 

87 """Dict-compatible metadata store that enforces extension contracts.""" 

88 

89 def __init__( 

90 self, 

91 scope_id: str | None, 

92 initial: Mapping[str, Any] | None = None, 

93 ) -> None: 

94 self.scope_id = scope_id 

95 super().__init__() 

96 if initial: 

97 self.update(initial) 

98 

99 def _validate_contracts( 

100 self, 

101 *, 

102 operation: str, 

103 metadata: Mapping[str, Any] | None = None, 

104 ) -> None: 

105 ObjectStateMetadataContractRegistry.validate_mapping( 

106 scope_id=self.scope_id, 

107 metadata=self if metadata is None else metadata, 

108 operation=operation, 

109 ) 

110 

111 def __setitem__(self, key: str, value: Any) -> None: 

112 ObjectStateMetadataContractRegistry.validate_key(key) 

113 super().__setitem__(key, value) 

114 

115 def update(self, other: Mapping[str, Any] | None = None, **kwargs: Any) -> None: 

116 if other: 

117 self._validate_contracts(operation="write", metadata=other) 

118 if kwargs: 

119 self._validate_contracts(operation="write", metadata=kwargs) 

120 super().update(other or {}, **kwargs) 

121 

122 def setdefault(self, key: str, default: Any = None) -> Any: 

123 ObjectStateMetadataContractRegistry.validate_key(key) 

124 return super().setdefault(key, default) 

125 

126 def copy_for_snapshot(self) -> dict[str, Any]: 

127 self._validate_contracts(operation="snapshot") 

128 return copy.deepcopy(dict(self)) 

129 

130 def __copy__(self) -> "ObjectStateMetadataStore": 

131 return ObjectStateMetadataStore(self.scope_id, dict(self)) 

132 

133 def __deepcopy__(self, memo: dict[int, Any]) -> "ObjectStateMetadataStore": 

134 copied = ObjectStateMetadataStore(self.scope_id) 

135 memo[id(self)] = copied 

136 copied.update(copy.deepcopy(dict(self), memo)) 

137 return copied 

138 

139 @classmethod 

140 def from_snapshot( 

141 cls, 

142 *, 

143 scope_id: str | None, 

144 metadata: Mapping[str, Any], 

145 ) -> "ObjectStateMetadataStore": 

146 ObjectStateMetadataContractRegistry.validate_mapping( 

147 scope_id=scope_id, 

148 metadata=metadata, 

149 operation="restore", 

150 ) 

151 return cls(scope_id, copy.deepcopy(dict(metadata)))