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

112 statements  

« prev     ^ index     » next       coverage.py v7.15.0, created at 2026-07-04 11:16 +0000

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 raise CyclicReferenceError(KeyInfo(Path([]), -1)) 

71 return 

72 seen.add(id(data)) 

73 

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

75 if not isinstance(k, str): 

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

77 raise TypeError(msg) 

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

79 if isinstance(v, Mapping): 

80 try: 

81 yield from ( 

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

83 ) 

84 except CyclicReferenceError as err: 

85 err.key_info = KeyInfo(Path((AttrKey(k), *err.key_info.path)), err.key_info.index + 1) 

86 raise 

87 if end_point_only: 

88 continue 

89 yield k 

90 seen.remove(id(data)) 

91 

92 

93@generate 

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

95 """ 

96 映射配置数据 

97 

98 .. versionadded:: 0.1.5 

99 """ 

100 

101 _data: D 

102 data: D 

103 

104 def __init__(self, data: D | None = None): # type: ignore[var-annotated] # mypy抽风 

105 """ 

106 :param data: 映射数据 

107 :type data: D | None 

108 """ # noqa: D205 

109 if data is None: 

110 data = {} 

111 if not isinstance(data, Mapping): 

112 msg = f"must be mapping, not {type(data).__name__}" 

113 raise TypeError(msg) 

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

115 

116 @property 

117 @override 

118 def data_read_only(self) -> bool: 

119 return not isinstance(self._data, MutableMapping) 

120 

121 @override 

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

123 # noinspection GrazieInspection 

124 r""" 

125 获取所有键 

126 

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

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

129 

130 :param recursive: 是否递归获取 

131 :type recursive: bool 

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

133 :type strict: bool 

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

135 :type end_point_only: bool 

136 

137 :return: 所有键 

138 :rtype: KeysView[str] 

139 

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

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

142 

143 例子 

144 ---- 

145 

146 >>> from c41811.config import MappingConfigData 

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

148 

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

150 

151 >>> data.keys() 

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

153 

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

155 

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

157 odict_keys(['foo1']) 

158 

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

160 

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

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

163 

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

165 

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

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

168 

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

170 

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

172 >>> cyclic["cyclic"] = cyclic 

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

174 

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

176 Traceback (most recent call last): 

177 ... 

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

179 

180 否则静默跳过循环引用 

181 

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

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

184 

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

186 odict_keys(['key']) 

187 

188 .. versionchanged:: 0.2.0 

189 添加参数 ``strict`` 

190 """ # noqa: RUF002 

191 if recursive: 

192 return OrderedDict.fromkeys( 

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

194 ).keys() 

195 

196 if end_point_only: 

197 return OrderedDict.fromkeys( 

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

199 ).keys() 

200 

201 return self._data.keys() 

202 

203 @override 

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

205 """ 

206 获取所有值 

207 

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

209 :type return_raw_value: bool 

210 

211 :return: 所有键值对 

212 :rtype: ValuesView[Any] 

213 

214 .. versionchanged:: 0.2.0 

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

216 """ 

217 if return_raw_value: 

218 return self._data.values() 

219 

220 return OrderedDict( 

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

222 ).values() 

223 

224 @override 

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

226 """ 

227 获取所有键值对 

228 

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

230 :type return_raw_value: bool 

231 

232 :return: 所有键值对 

233 :rtype: ItemsView[str, Any] 

234 

235 .. versionchanged:: 0.2.0 

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

237 """ 

238 if return_raw_value: 

239 return self._data.items() 

240 return OrderedDict( 

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

242 ).items() 

243 

244 @override 

245 @check_read_only 

246 def clear(self) -> None: 

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

248 

249 @override 

250 @check_read_only 

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

252 path = fmt_path(path) 

253 try: 

254 result = self.retrieve(path) 

255 self.delete(path) 

256 except RequiredPathNotFoundError: 

257 if default is not Unset: 

258 return default 

259 raise 

260 return result 

261 

262 @override 

263 @check_read_only 

264 def popitem(self) -> Any: 

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

266 

267 @override 

268 @check_read_only 

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

270 if m is not None: 

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

272 return 

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

274 

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

276 try: 

277 return self[item] 

278 except KeyError: 

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

280 raise AttributeError(msg) from None 

281 

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

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

284 ... 

285 

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

287 ... 

288 

289 

290__all__ = ("MappingConfigData",)