Coverage for src / c41811 / config / basic / mapping.py: 100%

1# cython: language_level = 3 # noqa: ERA001 

2 

3 

4""" 

5映射类型配置数据实现 

6 

7.. versionadded:: 0.2.0 

8""" 

9 

10import operator 

11from collections import OrderedDict 

12from collections.abc import Generator 

13from collections.abc import ItemsView 

14from collections.abc import KeysView 

15from collections.abc import Mapping 

16from collections.abc import MutableMapping 

17from collections.abc import ValuesView 

18from copy import deepcopy 

19from typing import Any 

20from typing import Self 

21from typing import cast 

22from typing import override 

23 

24from ._generate_operators import generate 

25from ._generate_operators import operate 

26from .core import BasicIndexedConfigData 

27from .utils import check_read_only 

28from .utils import fmt_path 

29from ..abc import PathLike 

30from ..errors import CyclicReferenceError 

31from ..errors import KeyInfo 

32from ..errors import RequiredPathNotFoundError 

33from ..path import AttrKey 

34from ..path import Path 

35from ..utils import Unset 

36 

37 

38def _keys_recursive( 

39 data: Mapping[Any, Any], 

40 seen: set[int] | None = None, 

41 *, 

42 strict: bool, 

43 end_point_only: bool, 

44) -> Generator[str, None, None]: 

45 """ 

46 递归获取配置的键 

47 

48 :param data: 配置数据 

49 :type data: Mapping 

50 :param seen: 已访问的配置数据的id 

51 :type seen: set[int] | None 

52 :param strict: 是否严格模式，如果为 True，则当遇到循环引用时，会抛出异常 

53 :type strict: bool 

54 :param end_point_only: 是否只返回叶子节点的键 

55 :type end_point_only: bool 

56 

57 :return: 获取的生成器 

58 :rtype: Generator[str, None, None] 

59 

60 :raises CyclicReferenceError: 当遇到循环引用时，如果 strict 为 True，则抛出此异常 

61 :raises TypeError: 递归获取时键不为str时抛出 

62 

63 .. versionadded:: 0.2.0 

64 """ # noqa: RUF002 

65 if seen is None: 

66 seen = set() 

67 

68 if id(data) in seen: 

69 if strict: 

70 # noinspection PyTypeChecker 

71 raise CyclicReferenceError(key_info=KeyInfo(Path([]), None, -1)) 

72 return 

73 seen.add(id(data)) 

74 

75 for k, v in data.items(): 

76 if not isinstance(k, str): 

77 msg = f"key must be str, not {type(k).__name__}" 

78 raise TypeError(msg) 

79 k = k.replace("\\", "\\\\") # noqa: PLW2901 

80 if isinstance(v, Mapping): 

81 try: 

82 yield from ( 

83 f"{k}\\.{x}" for x in _keys_recursive(v, seen, strict=strict, end_point_only=end_point_only) 

84 ) 

85 except CyclicReferenceError as err: 

86 key_info = err.key_info 

87 key = AttrKey(k) 

88 

89 key_info.path = Path((key, *key_info.path)) 

90 key_info.current_key = key if key_info.current_key is None else key_info.current_key 

91 key_info.index += 1 

92 raise 

93 if end_point_only: 

94 continue 

95 yield k 

96 seen.remove(id(data)) 

97 

98 

99@generate 

100class MappingConfigData[D: Mapping[Any, Any]](BasicIndexedConfigData[D], MutableMapping[Any, Any]): 

101 """ 

102 映射配置数据 

103 

104 .. versionadded:: 0.1.5 

105 """ 

106 

107 _data: D 

108 data: D 

109 

110 def __init__(self, data: D | None = None): 

111 """ 

112 :param data: 映射数据 

113 :type data: D | None 

114 """ # noqa: D205 

115 if data is None: 

116 data = {} # type: ignore[assignment] 

117 super().__init__(cast(D, data)) 

118 

119 @property 

120 @override 

121 def data_read_only(self) -> bool: 

122 return not isinstance(self._data, MutableMapping) 

123 

124 @override 

125 def keys(self, *, recursive: bool = False, strict: bool = True, end_point_only: bool = False) -> KeysView[Any]: 

126 # noinspection GrazieInspection 

127 r""" 

128 获取所有键 

129 

130 不为 :py:class:`~collections.abc.Mapping` 默认行为时键必须为 :py:class:`str` 且返回值会被转换为 

131 :ref:`配置数据路径字符串 <term-config-data-path-syntax>` 

132 

133 :param recursive: 是否递归获取 

134 :type recursive: bool 

135 :param strict: 是否严格检查循环引用数据，为真时提前抛出错误，否则静默忽略 

136 :type strict: bool 

137 :param end_point_only: 是否只获取叶子节点 

138 :type end_point_only: bool 

139 

140 :return: 所有键 

141 :rtype: KeysView[str] 

142 

143 :raise TypeError: 递归获取时键不为str时抛出 

144 :raise CyclicReferenceError: 严格检查循环引用数据时发现循环引用抛出 

145 

146 例子 

147 ---- 

148 

149 >>> from c41811.config import MappingConfigData 

150 >>> data = MappingConfigData({"foo": {"bar": {"baz": "value"}, "bar1": "value1"}, "foo1": "value2"}) 

151 

152 不带参数行为与普通字典一样 

153 

154 >>> data.keys() 

155 dict_keys(['foo', 'foo1']) 

156 

157 参数 ``end_point_only`` 会滤掉非 ``叶子节点`` 的键 

158 

159 >>> data.keys(end_point_only=True) # 内部计算为保留顺序采用了OrderedDict所以返回值是odict_keys 

160 odict_keys(['foo1']) 

161 

162 参数 ``recursive`` 用于获取所有的 ``路径`` 

163 

164 >>> data.keys(recursive=True) 

165 odict_keys(['foo\\.bar\\.baz', 'foo\\.bar', 'foo\\.bar1', 'foo', 'foo1']) 

166 

167 同时提供 ``recursive`` 和 ``end_point_only`` 会产出所有 ``叶子节点`` 的路径 

168 

169 >>> data.keys(recursive=True, end_point_only=True) 

170 odict_keys(['foo\\.bar\\.baz', 'foo\\.bar1', 'foo1']) 

171 

172 为严格模式时会检查循环引用并提前引发错误 

173 

174 >>> cyclic: dict[str, Any] = {"cyclic": None, "key": "value"} 

175 >>> cyclic["cyclic"] = cyclic 

176 >>> cyclic: MappingConfigData[dict[str, Any]] = MappingConfigData(cyclic) 

177 

178 >>> cyclic.keys(recursive=True) # 默认为严格模式 

179 Traceback (most recent call last): 

180 ... 

181 c41811.config.errors.CyclicReferenceError: Cyclic reference detected at \.cyclic -> \.cyclic (1/1) 

182 

183 否则静默跳过循环引用 

184 

185 >>> cyclic.keys(recursive=True, strict=False) 

186 odict_keys(['cyclic', 'key']) 

187 

188 >>> cyclic.keys(recursive=True, strict=False, end_point_only=True) 

189 odict_keys(['key']) 

190 

191 .. versionchanged:: 0.2.0 

192 添加参数 ``strict`` 

193 """ # noqa: RUF002 

194 if recursive: 

195 return OrderedDict.fromkeys( 

196 x for x in _keys_recursive(self._data, strict=strict, end_point_only=end_point_only) 

197 ).keys() 

198 

199 if end_point_only: 

200 return OrderedDict.fromkeys( 

201 k.replace("\\", "\\\\") for k, v in self._data.items() if not isinstance(v, Mapping) 

202 ).keys() 

203 

204 return self._data.keys() 

205 

206 @override 

207 def values(self, return_raw_value: bool = False) -> ValuesView[Any]: 

208 """ 

209 获取所有值 

210 

211 :param return_raw_value: 是否获取原始数据 

212 :type return_raw_value: bool 

213 

214 :return: 所有键值对 

215 :rtype: ValuesView[Any] 

216 

217 .. versionchanged:: 0.2.0 

218 重命名参数 ``get_raw`` 为 ``return_raw_value`` 

219 """ 

220 if return_raw_value: 

221 return self._data.values() 

222 

223 return OrderedDict( 

224 (k, self.from_data(v) if isinstance(v, Mapping) else deepcopy(v)) for k, v in self._data.items() 

225 ).values() 

226 

227 @override 

228 def items(self, *, return_raw_value: bool = False) -> ItemsView[str, Any]: 

229 """ 

230 获取所有键值对 

231 

232 :param return_raw_value: 是否获取原始数据 

233 :type return_raw_value: bool 

234 

235 :return: 所有键值对 

236 :rtype: ItemsView[str, Any] 

237 

238 .. versionchanged:: 0.2.0 

239 重命名参数 ``get_raw`` 为 ``return_raw_value`` 

240 """ 

241 if return_raw_value: 

242 return self._data.items() 

243 return OrderedDict( 

244 (deepcopy(k), self.from_data(v) if isinstance(v, Mapping) else deepcopy(v)) for k, v in self._data.items() 

245 ).items() 

246 

247 @override 

248 @check_read_only 

249 def clear(self) -> None: 

250 self._data.clear() # type: ignore[attr-defined] 

251 

252 @override 

253 @check_read_only 

254 def pop(self, path: PathLike, /, default: Any = Unset) -> Any: 

255 path = fmt_path(path) 

256 try: 

257 result = self.retrieve(path) 

258 self.delete(path) 

259 except RequiredPathNotFoundError: 

260 if default is not Unset: 

261 return default 

262 raise 

263 return result 

264 

265 @override 

266 @check_read_only 

267 def popitem(self) -> Any: 

268 return self._data.popitem() # type: ignore[attr-defined] 

269 

270 @override 

271 @check_read_only 

272 def update(self, m: Any = None, /, **kwargs: Any) -> None: 

273 if m is not None: 

274 self._data.update(m) # type: ignore[attr-defined] 

275 return 

276 self._data.update(**kwargs) # type: ignore[attr-defined] 

277 

278 def __getattr__(self, item: Any) -> Self | Any: 

279 try: 

280 return self[item] 

281 except KeyError: 

282 msg = f"'{self.__class__.__name__}' object has no attribute '{item}'" 

283 raise AttributeError(msg) from None 

284 

285 @operate(operator.or_, operator.ior) 

286 def __or__(self, other: Any) -> Self: # type: ignore[empty-body] 

287 ... 

288 

289 def __ror__(self, other: Any) -> Self: # type: ignore[empty-body] 

290 ... 

291 

292 

293__all__ = ("MappingConfigData",)