Coverage for src/c41811/config/path.py: 100%

202 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 

6import warnings 

7from abc import ABC 

8from collections.abc import Iterable 

9from collections.abc import Mapping 

10from collections.abc import MutableMapping 

11from collections.abc import MutableSequence 

12from collections.abc import Sequence 

13from functools import lru_cache 

14from typing import Any 

15from typing import Self 

16from typing import cast 

17from typing import override 

18 

19from ._protocols import Indexed 

20from .abc import ABCKey 

21from .abc import ABCPath 

22from .errors import ConfigDataPathSyntaxError 

23from .errors import ConfigDataPathSyntaxWarning 

24from .errors import TokenInfo 

25 

26 

27class IndexMixin[K, D: Indexed[Any, Any]](ABCKey[K, D], ABC): 

28 """ 

29 混入类,提供对Index操作的支持 

30 

31 .. versionchanged:: 0.1.5 

32 重命名 ``ItemMixin`` 为 ``IndexMixin`` 

33 """ # noqa: RUF002 

34 

35 @override 

36 def __get_inner_element__(self, data: D) -> D: 

37 return cast(D, data[self._key]) 

38 

39 @override 

40 def __set_inner_element__(self, data: D, value: Any) -> None: 

41 data[self._key] = value # type: ignore[index] 

42 

43 @override 

44 def __delete_inner_element__(self, data: D) -> None: 

45 del data[self._key] # type: ignore[attr-defined] 

46 

47 

48class AttrKey(IndexMixin[str, Mapping[str, Any]], ABCKey[str, Mapping[str, Any]]): 

49 """属性键""" 

50 

51 _key: str 

52 

53 def __init__(self, key: str, meta: str | None = None): 

54 """ 

55 :param key: 键名 

56 :type key: str 

57 :param meta: 元信息 

58 :type meta: str | None 

59 

60 :raise TypeError: key不为str时抛出 

61 """ # noqa: D205 

62 if not isinstance(key, str): 

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

64 raise TypeError(msg) 

65 super().__init__(key, meta) 

66 

67 @override 

68 def __contains_inner_element__(self, data: Mapping[Any, Any]) -> bool: 

69 return self._key in data 

70 

71 @override 

72 def __supports__(self, data: Any) -> tuple[Any, ...]: 

73 return () if isinstance(data, Mapping) else (Mapping,) 

74 

75 @override 

76 def __supports_modify__(self, data: Any) -> tuple[Any, ...]: 

77 return () if isinstance(data, MutableMapping) else (MutableMapping,) 

78 

79 @override 

80 def unparse(self) -> str: 

81 meta = "" if self._meta is None else f"\\{{{self._meta.replace('\\', '\\\\')}\\}}" 

82 return f"{meta}\\.{self._key.replace('\\', '\\\\')}" 

83 

84 def __len__(self) -> int: 

85 return len(self._key) 

86 

87 @override 

88 def __eq__(self, other: Any) -> bool: 

89 if isinstance(other, str): 

90 return self._key == other 

91 return super().__eq__(other) 

92 

93 @override 

94 def __hash__(self) -> int: 

95 return super().__hash__() 

96 

97 

98class IndexKey(IndexMixin[int, Sequence[Any]], ABCKey[int, Sequence[Any]]): 

99 """下标键""" 

100 

101 _key: int 

102 

103 def __init__(self, key: int, meta: str | None = None): 

104 """ 

105 :param key: 索引值 

106 :type key: int 

107 :param meta: 元信息 

108 :type meta: str 

109 

110 :raise TypeError: key不为int时抛出 

111 """ # noqa: D205 

112 if not isinstance(key, int): 

113 msg = f"key must be int, not {type(key).__name__}" 

114 raise TypeError(msg) 

115 super().__init__(key, meta) 

116 

117 @override 

118 def __contains_inner_element__(self, data: Sequence[Any]) -> bool: 

119 try: 

120 data[self._key] 

121 except IndexError: 

122 return False 

123 return True 

124 

125 @override 

126 def __supports__(self, data: Any) -> tuple[Any, ...]: 

127 return () if isinstance(data, Sequence) else (Sequence,) 

128 

129 @override 

130 def __supports_modify__(self, data: Any) -> tuple[Any, ...]: 

131 return () if isinstance(data, MutableSequence) else (MutableSequence,) 

132 

133 @override 

134 def unparse(self) -> str: 

135 meta = "" if self._meta is None else f"\\{{{self._meta.replace('\\', '\\\\')}\\}}" 

136 return f"{meta}\\[{self._key}\\]" 

137 

138 

139class Path(ABCPath[AttrKey | IndexKey]): 

140 """配置数据路径""" 

141 

142 @classmethod 

143 def from_str(cls, string: str) -> Self: 

144 """ 

145 从字符串解析路径 

146 

147 :param string: 路径字符串 

148 :type string: str 

149 

150 :return: 解析后的路径 

151 :rtype: Self 

152 """ 

153 return cls(PathSyntaxParser.parse(string)) 

154 

155 @classmethod 

156 def from_locate(cls, locate: Iterable[str | int]) -> Self: 

157 """ 

158 从列表解析路径 

159 

160 :param locate: 键列表 

161 :type locate: Iterable[str | int] 

162 

163 :return: 解析后的路径 

164 :rtype: Self 

165 """ 

166 keys: list[AttrKey | IndexKey] = [] 

167 for loc in locate: 

168 if isinstance(loc, int): 

169 keys.append(IndexKey(loc)) 

170 continue 

171 if isinstance(loc, str): 

172 keys.append(AttrKey(loc)) 

173 continue 

174 msg = "locate element must be 'int' or 'str'" 

175 raise ValueError(msg) 

176 return cls(keys) 

177 

178 def to_locate(self) -> list[str | int]: 

179 """ 

180 转换为列表 

181 

182 .. versionadded:: 0.1.1 

183 """ 

184 return [key.key for key in self._keys] 

185 

186 @override 

187 def unparse(self) -> str: 

188 return "".join(key.unparse() for key in self._keys) 

189 

190 

191def _count_backslash(s: str) -> int: 

192 count = 1 

193 while s and (s[-1] == "\\"): 

194 count += 1 

195 s = s[:-1] 

196 return count 

197 

198 

199class PathSyntaxParser: 

200 """路径语法解析器""" 

201 

202 @staticmethod 

203 @lru_cache 

204 def tokenize(string: str) -> tuple[str, ...]: # noqa: C901 (ignore complexity) 

205 # noinspection GrazieInspection 

206 r""" 

207 将字符串分词为以\开头的有意义片段 

208 

209 :param string: 待分词字符串 

210 :type string: str 

211 

212 :return: 分词结果 

213 :rtype: tuple[str, ...] 

214 

215 .. note:: 

216 可以省略字符串开头的 ``\.`` 

217 

218 例如: 

219 

220 ``r"\.first\.second\.third“`` 

221 

222 可以简写为 

223 

224 ``r"first\.second\.third"`` 

225 

226 .. versionchanged:: 0.1.4 

227 允许省略字符串开头的 ``\.`` 

228 

229 更改返回值类型 ``Generator[str, None, None]`` 为 ``tuple[str, ...]`` 

230 

231 添加缓存 

232 """ # noqa: RUF002 

233 # 开头默认为AttrKey 

234 if not string.startswith((r"\.", r"\[", r"\{")): 

235 string = rf"\.{string}" 

236 

237 tokens: list[str] = [""] 

238 invalid_escape_indexes: dict[int, list[str]] = {} 

239 while string: 

240 string, sep, token = string.rpartition("\\") 

241 

242 invalid_escape = False 

243 # 处理r"\\"防止转义 

244 if not token: 

245 token += tokens.pop() 

246 # 对不存在的转义进行警告 # 检查这个转义符号是否已经被转义 

247 elif (token[0] not in {".", "\\", "[", "]", "{", "}"}) and _count_backslash(string) % 2: 

248 invalid_escape = True 

249 

250 # 连接不应单独存在的token 

251 index_safe = tokens and (len(tokens[-1]) > 1) 

252 if index_safe and (tokens[-1][1] not in {".", "[", "]", "{", "}"}): 

253 token += tokens.pop() 

254 if invalid_escape: 

255 invalid_escape_indexes.setdefault(len(tokens) - 1 if "" in tokens else len(tokens), []).append( 

256 f"\\{token[0]}" 

257 ) 

258 

259 # 将 r"\]" 和 r"\}" 后面紧随的字符单独切割出来 

260 if token.startswith(("]", "}")) and token[1:]: 

261 tokens.append(token[1:]) 

262 token = token[:1] 

263 

264 tokens.append(sep + token) 

265 

266 tokens.reverse() 

267 if tokens[-1] == "": 

268 tokens.pop() 

269 

270 final_tokens = tuple(tokens) 

271 

272 for invalid_token_index, invalid_sequences in invalid_escape_indexes.items(): 

273 if len(invalid_sequences) > 1: 

274 msg = f"Invalid escape sequences ('{"', '".join(invalid_sequences)}')" 

275 else: 

276 msg = f"Invalid escape sequence ('{invalid_sequences[0]}')" 

277 warnings.warn( 

278 ConfigDataPathSyntaxWarning(msg, TokenInfo(final_tokens, invalid_token_index * -1 - 1)), stacklevel=2 

279 ) 

280 

281 return final_tokens 

282 

283 @classmethod 

284 @lru_cache 

285 def parse(cls, string: str) -> tuple[AttrKey | IndexKey, ...]: # noqa: C901 (ignore complexity) 

286 """ 

287 解析字符串为键列表 

288 

289 :param string: 待解析字符串 

290 :type string: str 

291 

292 :return: 键列表 

293 :rtype: list[AttrKey | IndexKey] 

294 

295 .. versionchanged:: 0.3.2 

296 更改返回值类型 ``list[AttrKey | IndexKey]`` 为 ``tuple[AttrKey | IndexKey, ...]`` 

297 

298 添加缓存 

299 """ 

300 path: list[AttrKey | IndexKey] = [] 

301 item: str | None = None 

302 meta: str | None = None 

303 token_stack: list[str] = [] 

304 

305 tokenized_path = cls.tokenize(string) 

306 for index, token in enumerate(tokenized_path): 

307 if not token.startswith("\\"): 

308 msg = r"Unexpected token. Did you missed `\.` `\[` or `\{` before it?" 

309 raise ConfigDataPathSyntaxError(msg, TokenInfo(tokenized_path, index)) 

310 

311 token_type = token[1] 

312 content = token[2:].replace("\\\\", "\\") 

313 

314 def _token_closed(tk_typ: str, tk_close: str, i: int) -> None: 

315 try: 

316 top = token_stack.pop() 

317 except IndexError: 

318 m = f"unmatched '{tk_close}'" 

319 raise ConfigDataPathSyntaxError(m, TokenInfo(tokenized_path, i)) from None 

320 if top != tk_typ: 

321 m = f"closing parenthesis '{tk_close}' does not match opening parenthesis '{top}'" 

322 raise ConfigDataPathSyntaxError(m, TokenInfo(tokenized_path, i)) 

323 

324 if token_type == "}": # noqa: S105 

325 _token_closed("{", "}", index) 

326 continue 

327 if token_type == "]": # noqa: S105 

328 _token_closed("[", "]", index) 

329 try: 

330 path.append(IndexKey(int(item), meta)) # type: ignore[arg-type] 

331 except ValueError: 

332 msg = f"index key '{item}' must be numeric" 

333 raise ConfigDataPathSyntaxError(msg, TokenInfo(tokenized_path, index)) from None 

334 item = None 

335 meta = None 

336 continue 

337 

338 if token_stack: 

339 msg = f"'{token_stack.pop()}' was never closed" 

340 raise ConfigDataPathSyntaxError(msg, TokenInfo(tokenized_path, index)) 

341 

342 if token_type == "[": # noqa: S105 

343 token_stack.append("[") 

344 item = content 

345 continue 

346 if token_type == "{": # noqa: S105 

347 token_stack.append("{") 

348 meta = content 

349 continue 

350 if token_type == ".": # noqa: S105 

351 path.append(AttrKey(content, meta)) 

352 meta = None 

353 continue 

354 

355 msg = r"Unexpected token. Did you missed `\.` `\[` or `\{` before it?" 

356 raise ConfigDataPathSyntaxError(msg, TokenInfo(tokenized_path, index)) 

357 

358 if token_stack: 

359 msg = f"'{token_stack.pop()}' was never closed" 

360 raise ConfigDataPathSyntaxError(msg, TokenInfo(tokenized_path, -1)) 

361 

362 if meta: 

363 warnings.warn( 

364 ConfigDataPathSyntaxWarning(r"Isolate meta found", TokenInfo(tokenized_path, -1)), stacklevel=2 

365 ) 

366 

367 return tuple(path) 

368 

369 

370__all__ = ( 

371 "AttrKey", 

372 "IndexKey", 

373 "Path", 

374 "PathSyntaxParser", 

375)