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

1# cython: language_level = 3 # noqa: ERA001 

2 

3 

4""" 

5组件配置数据实现 

6 

7.. versionadded:: 0.2.0 

8""" 

9 

10from collections.abc import Callable 

11from collections.abc import Iterator 

12from collections.abc import Mapping 

13from collections.abc import MutableMapping 

14from contextlib import suppress 

15from copy import deepcopy 

16from dataclasses import dataclass 

17from dataclasses import field 

18from typing import Any 

19from typing import Self 

20from typing import cast 

21from typing import override 

22 

23from .core import BasicConfigData 

24from .factory import ConfigDataFactory 

25from .utils import check_read_only 

26from .utils import fmt_path 

27from ..abc import ABCConfigData 

28from ..abc import ABCIndexedConfigData 

29from ..abc import ABCMetaParser 

30from ..abc import ABCPath 

31from ..abc import PathLike 

32from ..errors import ComponentMemberMismatchError 

33from ..errors import ComponentMetadataException 

34from ..errors import ConfigDataTypeError 

35from ..errors import ConfigOperate 

36from ..errors import KeyInfo 

37from ..errors import RequiredPathNotFoundError 

38 

39 

40@dataclass 

41class ComponentOrders: 

42 """ 

43 组件顺序 

44 

45 .. versionadded:: 0.2.0 

46 """ 

47 

48 create: list[str] = field(default_factory=list) 

49 read: list[str] = field(default_factory=list) 

50 update: list[str] = field(default_factory=list) 

51 delete: list[str] = field(default_factory=list) 

52 

53 

54@dataclass 

55class ComponentMember: 

56 """ 

57 组件成员 

58 

59 .. versionadded:: 0.2.0 

60 """ 

61 

62 filename: str 

63 alias: str | None = field(default=None) 

64 config_format: str | None = field(default=None) 

65 

66 

67@dataclass 

68class ComponentMeta[D: ABCConfigData]: 

69 """ 

70 组件元数据 

71 

72 .. versionadded:: 0.2.0 

73 """ 

74 

75 config: D = field(default_factory=ConfigDataFactory) # type: ignore[assignment] 

76 orders: ComponentOrders = field(default_factory=ComponentOrders) 

77 members: list[ComponentMember] = field(default_factory=list) 

78 parser: ABCMetaParser[Any, Any] | None = field(default=None) 

79 

80 

81class ComponentConfigData[D: ABCIndexedConfigData[Any], M: ComponentMeta[Any]]( 

82 BasicConfigData[D], ABCIndexedConfigData[D] 

83): 

84 """ 

85 组件配置数据 

86 

87 .. versionadded:: 0.2.0 

88 """ 

89 

90 def __init__(self, meta: M | None = None, members: Mapping[str, D] | None = None): 

91 """ 

92 :param meta: 组件元数据 

93 :type meta: M | None 

94 :param members: 组件成员 

95 :type members: Mapping[str, D] | None 

96 """ # noqa: D205 

97 if meta is None: 

98 meta = ComponentMeta() # type: ignore[assignment] 

99 if members is None: 

100 members = {} 

101 

102 # 准备元数据 

103 self._meta: M = cast(M, deepcopy(meta)) 

104 self._filename2meta: dict[str, ComponentMember] = {} 

105 self._alias2filename: dict[str, str] = {} 

106 for member_meta in self._meta.members: 

107 if member_meta.filename in self._filename2meta: # 文件名不能重复 

108 msg = f"filename {member_meta.filename} is repeated" 

109 raise ComponentMetadataException(msg) 

110 self._filename2meta[member_meta.filename] = member_meta 

111 if member_meta.filename in self._alias2filename: # 别名不能和文件名重复 

112 msg = f"alias {member_meta.filename} is same as filename {member_meta.filename}" 

113 raise ComponentMetadataException(msg) 

114 if member_meta.alias is None: 

115 continue 

116 if member_meta.alias in self._alias2filename: # 别名不能重复 

117 msg = f"alias {member_meta.alias} is repeated" 

118 raise ComponentMetadataException(msg) 

119 if member_meta.alias in self._filename2meta: # 别名不能和文件名相同 

120 msg = f"alias {member_meta.alias} is same as filename {member_meta.filename}" 

121 raise ComponentMetadataException(msg) 

122 self._alias2filename[member_meta.alias] = member_meta.filename 

123 

124 self._members: Mapping[str, D] = deepcopy(members) 

125 missing = self._filename2meta.keys() - self._members.keys() 

126 redundant = self._members.keys() - self._filename2meta.keys() 

127 if missing | redundant: 

128 raise ComponentMemberMismatchError(missing=missing, redundant=redundant) 

129 

130 @property 

131 def meta(self) -> M: 

132 """ 

133 组件元信息 

134 

135 .. caution:: 

136 未默认做深拷贝，可能导致非预期行为 

137 

138 除非你知道你在做什么，不要轻易修改！ 

139 

140 由于 :py:class:`ComponentMeta` 仅提供一个通用的接口， 

141 直接修改其中元数据而不修改 ``config`` 字段 `*可能*` 会导致SL与元数据的不同步， 

142 这取决于 :py:class:`ComponentSL` 所取用的元数据解析器的行为 

143 """ # noqa: RUF002 

144 return self._meta 

145 

146 @property 

147 def members(self) -> Mapping[str, D]: 

148 """ 

149 组件成员 

150 

151 .. caution:: 

152 未默认做深拷贝，可能导致非预期行为 

153 """ # noqa: RUF002 

154 return self._members 

155 

156 @property 

157 @override 

158 def data_read_only(self) -> bool | None: 

159 """组件数据是否为只读""" 

160 return not isinstance(self._members, MutableMapping) 

161 

162 @property 

163 def filename2meta(self) -> Mapping[str, ComponentMember]: 

164 """文件名到成员元信息的映射""" 

165 return deepcopy(self._filename2meta) 

166 

167 @property 

168 def alias2filename(self) -> Mapping[str, str]: 

169 """别名到文件名的映射""" 

170 return deepcopy(self._alias2filename) 

171 

172 def _member(self, member: str) -> D: 

173 """ 

174 通过成员文件名以及其别名获取成员配置数据 

175 

176 :param member: 成员名 

177 :type member: str 

178 

179 :return: 成员数据 

180 :rtype: D 

181 """ 

182 try: 

183 return self._members[member] 

184 except KeyError: 

185 with suppress(KeyError): 

186 return self._members[self._alias2filename[member]] 

187 raise 

188 

189 def _resolve_members[P: ABCPath[Any], R]( 

190 self, path: P, order: list[str], processor: Callable[[P, D], R], exception: Exception 

191 ) -> R: 

192 """ 

193 逐个尝试解析成员配置数据 

194 

195 :param path: 路径 

196 :type path: P 

197 :param order: 成员处理顺序 

198 :type order: list[str] 

199 :param processor: 成员处理函数 

200 :type processor: Callable[[P, D], R] 

201 :param exception: 顺序为空抛出的错误 

202 :type exception: Exception 

203 

204 :return: 处理结果 

205 :rtype: R 

206 

207 .. important:: 

208 针对 :py:exc:`RequiredPathNotFoundError` ， :py:exc:`ConfigDataTypeError` 做了特殊处理， 

209 多个成员都抛出其一时最终仅抛出其中 :py:attr:`KeyInfo.index` 最大的 

210 """ # noqa: RUF002 

211 if path and (path[0].meta is not None): 

212 try: 

213 selected_member = self._member(path[0].meta) 

214 except KeyError: 

215 raise exception from None 

216 return processor(path, selected_member) 

217 

218 if not order: 

219 raise exception 

220 

221 error: RequiredPathNotFoundError | ConfigDataTypeError | None = None 

222 for member in order: 

223 try: 

224 return processor(path, self._member(member)) 

225 except (RequiredPathNotFoundError, ConfigDataTypeError) as err: 

226 if error is None: 

227 error = err 

228 if err.key_info.index > error.key_info.index: 

229 error = err 

230 raise cast(RequiredPathNotFoundError | ConfigDataTypeError, error) from None 

231 

232 @override 

233 def retrieve(self, path: PathLike, *args: Any, **kwargs: Any) -> Any: 

234 path = fmt_path(path) 

235 

236 def processor(pth: ABCPath[Any], member: D) -> Any: 

237 return member.retrieve(pth, *args, **kwargs) 

238 

239 return self._resolve_members( 

240 path, 

241 order=self._meta.orders.read, 

242 processor=processor, 

243 exception=RequiredPathNotFoundError( 

244 key_info=KeyInfo(path, path[0], 0), 

245 operate=ConfigOperate.Read, 

246 ), 

247 ) 

248 

249 @override 

250 @check_read_only 

251 def modify(self, path: PathLike, *args: Any, **kwargs: Any) -> Self: 

252 # noinspection PyIncorrectDocstring 

253 """ 

254 修改路径的值 

255 

256 :param path: 路径 

257 :type path: PathLike 

258 :param value: 值 

259 :type value: Any 

260 :param allow_create: 是否允许创建不存在的路径，默认为True 

261 :type allow_create: bool 

262 

263 :return: 返回当前实例便于链式调用 

264 :rtype: Self 

265 

266 :raise ConfigDataReadOnlyError: 配置数据为只读 

267 :raise ConfigDataTypeError: 配置数据类型错误 

268 :raise RequiredPathNotFoundError: 需求的键不存在 

269 

270 .. caution:: 

271 ``value`` 参数未默认做深拷贝，可能导致非预期行为 

272 

273 .. attention:: 

274 ``allow_create`` 时，使用与 `self.data` 一样的类型新建路径 

275 

276 .. versionchanged:: 0.3.0 

277 现在正确的先尝试使用 :py:attr:`~ComponentOrders.update` 对现有数据进行更新再尝试通过 

278 :py:attr:`~ComponentOrders.create` 创建新数据 

279 """ # noqa: RUF002 

280 path = fmt_path(path) 

281 

282 def _update_processor(pth: ABCPath[Any], member: D) -> None: 

283 try: 

284 member.retrieve(pth, return_raw_value=True) # 避免转换返回值带来的额外开销 

285 except (RequiredPathNotFoundError, ConfigDataTypeError) as err: 

286 raise RequiredPathNotFoundError( 

287 key_info=err.key_info, 

288 operate=ConfigOperate.Write, # 将操作从Read变为Write 

289 ) from None 

290 member.modify(pth, *args, **kwargs) 

291 

292 with suppress(RequiredPathNotFoundError): 

293 self._resolve_members( 

294 path, 

295 order=self._meta.orders.update, 

296 processor=_update_processor, 

297 exception=RequiredPathNotFoundError( 

298 key_info=KeyInfo(path, path[0], 0), 

299 operate=ConfigOperate.Write, 

300 ), 

301 ) 

302 return self 

303 

304 def _create_processor(pth: ABCPath[Any], member: D) -> None: 

305 member.modify(pth, *args, **kwargs) 

306 

307 self._resolve_members( 

308 path, 

309 order=self._meta.orders.create, 

310 processor=_create_processor, 

311 exception=RequiredPathNotFoundError( 

312 key_info=KeyInfo(path, path[0], 0), 

313 operate=ConfigOperate.Write, 

314 ), 

315 ) 

316 return self 

317 

318 @override 

319 @check_read_only 

320 def delete(self, path: PathLike, *args: Any, **kwargs: Any) -> Self: 

321 path = fmt_path(path) 

322 

323 def processor(pth: ABCPath[Any], member: D) -> None: 

324 # noinspection PyArgumentList 

325 member.delete(pth, *args, **kwargs) 

326 

327 self._resolve_members( 

328 path, 

329 order=self._meta.orders.delete, 

330 processor=processor, 

331 exception=RequiredPathNotFoundError( 

332 key_info=KeyInfo(path, path[0], 0), 

333 operate=ConfigOperate.Delete, 

334 ), 

335 ) 

336 return self 

337 

338 @override 

339 @check_read_only 

340 def unset(self, path: PathLike, *args: Any, **kwargs: Any) -> Self: 

341 path = fmt_path(path) 

342 

343 def processor(pth: ABCPath[Any], member: D) -> None: 

344 # noinspection PyArgumentList 

345 member.delete(pth, *args, **kwargs) 

346 

347 with suppress(RequiredPathNotFoundError): 

348 self._resolve_members( 

349 path, 

350 order=self._meta.orders.delete, 

351 processor=processor, 

352 exception=RequiredPathNotFoundError( 

353 key_info=KeyInfo(path, path[0], 0), 

354 operate=ConfigOperate.Delete, 

355 ), 

356 ) 

357 return self 

358 

359 @override 

360 def exists(self, path: PathLike, *args: Any, **kwargs: Any) -> bool: 

361 if not self._meta.orders.read: 

362 return False 

363 path = fmt_path(path) 

364 

365 def processor(pth: ABCPath[Any], member: D) -> bool: 

366 return member.exists(pth, *args, **kwargs) 

367 

368 with suppress(RequiredPathNotFoundError): # 个别极端条件触发 例如\{不存在的成员\}\.key 

369 return self._resolve_members( 

370 path, 

371 order=self._meta.orders.read, 

372 processor=processor, 

373 exception=RequiredPathNotFoundError( 

374 key_info=KeyInfo(path, path[0], 0), 

375 operate=ConfigOperate.Delete, 

376 ), 

377 ) 

378 return False 

379 

380 @override 

381 def get[V]( 

382 self, path: PathLike, default: V | None = None, *args: Any, return_raw_value: bool = False, **kwargs: Any 

383 ) -> V | Any: 

384 path = fmt_path(path) 

385 

386 def processor(pth: ABCPath[Any], member: D) -> Any: 

387 return member.retrieve(pth, *args, **kwargs) 

388 

389 with suppress(RequiredPathNotFoundError): 

390 return self._resolve_members( 

391 path, 

392 order=self._meta.orders.read, 

393 processor=processor, 

394 exception=RequiredPathNotFoundError( 

395 key_info=KeyInfo(path, path[0], 0), 

396 operate=ConfigOperate.Read, 

397 ), 

398 ) 

399 return default 

400 

401 @override 

402 @check_read_only 

403 def setdefault[V]( 

404 self, path: PathLike, default: V | None = None, *args: Any, return_raw_value: bool = False, **kwargs: Any 

405 ) -> V | Any: 

406 path = fmt_path(path) 

407 

408 def _retrieve_processor(pth: ABCPath[Any], member: D) -> Any: 

409 return member.retrieve(pth, *args, **kwargs) 

410 

411 with suppress(RequiredPathNotFoundError): 

412 return self._resolve_members( 

413 path, 

414 order=self._meta.orders.read, 

415 processor=_retrieve_processor, 

416 exception=RequiredPathNotFoundError( 

417 key_info=KeyInfo(path, path[0], 0), 

418 operate=ConfigOperate.Read, 

419 ), 

420 ) 

421 

422 def _modify_processor(pth: ABCPath[Any], member: D) -> Any: 

423 member.modify(pth, default) 

424 return default 

425 

426 return self._resolve_members( 

427 path, 

428 order=self._meta.orders.create, 

429 processor=_modify_processor, 

430 exception=RequiredPathNotFoundError( 

431 key_info=KeyInfo(path, path[0], 0), 

432 operate=ConfigOperate.Write, 

433 ), 

434 ) 

435 

436 @override 

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

438 if not isinstance(other, type(self)): 

439 return NotImplemented 

440 return all((self._meta == other._meta, self._members == other._members)) 

441 

442 __hash__ = None # type: ignore[assignment] 

443 

444 @override 

445 def __str__(self) -> str: 

446 return str(self._members) 

447 

448 @override 

449 def __repr__(self) -> str: 

450 return f"{self.__class__.__name__}(meta={self._meta!r}, members={self._members!r})" 

451 

452 def __deepcopy__(self, memo: dict[str, Any]) -> Self: 

453 return self.from_data(self._meta, self._members) 

454 

455 @override 

456 def __contains__(self, key: Any) -> bool: 

457 return key in self._members 

458 

459 @override 

460 def __iter__(self) -> Iterator[str]: 

461 return iter(self._members) 

462 

463 @override 

464 def __len__(self) -> int: 

465 return len(self._members) 

466 

467 @override 

468 def __getitem__(self, index: Any) -> D: 

469 return self._members[index] 

470 

471 @override 

472 @check_read_only 

473 def __setitem__(self, index: Any, value: D) -> None: 

474 """ 

475 .. danger:: 

476 使用此操作可能会导致与元数据不同步且不经过校验！ 

477 """ # noqa: RUF002, D205 

478 self._members[index] = value # type: ignore[index] 

479 

480 @override 

481 @check_read_only 

482 def __delitem__(self, index: Any) -> None: 

483 """ 

484 .. danger:: 

485 使用此操作可能会导致与元数据不同步且不经过校验！ 

486 """ # noqa: RUF002, D205 

487 del self._members[index] # type: ignore[attr-defined] 

488 

489 

490__all__ = ( 

491 "ComponentConfigData", 

492 "ComponentMember", 

493 "ComponentMeta", 

494 "ComponentOrders", 

495)