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

357 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 

10from abc import ABC 

11from collections import OrderedDict 

12from collections.abc import Callable 

13from collections.abc import Iterable 

14from collections.abc import Iterator 

15from collections.abc import Mapping 

16from collections.abc import Sequence 

17from contextlib import suppress 

18from copy import deepcopy 

19from re import Pattern 

20from typing import Any 

21from typing import Literal 

22from typing import Self 

23from typing import cast 

24from typing import overload 

25from typing import override 

26 

27from .factory import ConfigDataFactory 

28from .utils import check_read_only 

29from .utils import fmt_path 

30from .._protocols import Indexed 

31from ..abc import ABCConfigData 

32from ..abc import ABCConfigFile 

33from ..abc import ABCConfigPool 

34from ..abc import ABCIndexedConfigData 

35from ..abc import ABCPath 

36from ..abc import ABCProcessorHelper 

37from ..abc import ABCSLProcessorPool 

38from ..abc import AnyKey 

39from ..abc import PathLike 

40from ..errors import ConfigDataReadOnlyError 

41from ..errors import ConfigDataTypeError 

42from ..errors import ConfigOperate 

43from ..errors import FailedProcessConfigFileError 

44from ..errors import KeyInfo 

45from ..errors import RequiredPathNotFoundError 

46from ..errors import UnsupportedConfigFormatError 

47 

48 

49class BasicConfigData[D](ABCConfigData, ABC): 

50 # noinspection GrazieInspection 

51 """ 

52 配置数据基类 

53 

54 .. versionadded:: 0.1.5 

55 

56 .. versionchanged:: 0.2.0 

57 重命名 ``BaseConfigData`` 为 ``BasicConfigData`` 

58 """ 

59 

60 _read_only: bool | None = False 

61 

62 @property 

63 @override 

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

65 return True # 全被子类复写了 测不到 # pragma: no cover 

66 

67 @property # type: ignore[explicit-override] # mypy抽风 

68 @override 

69 def read_only(self) -> bool | None: 

70 return super().read_only or self._read_only 

71 

72 @read_only.setter 

73 @override 

74 def read_only(self, value: Any) -> None: 

75 if self.data_read_only: 

76 raise ConfigDataReadOnlyError 

77 self._read_only = bool(value) 

78 

79 

80class BasicSingleConfigData[D](BasicConfigData[D], ABC): 

81 """ 

82 单文件配置数据基类 

83 

84 .. versionadded:: 0.2.0 

85 """ 

86 

87 def __init__(self, data: D): 

88 """ 

89 :param data: 配置的原始数据 

90 :type data: Any 

91 """ # noqa: D205 

92 self._data: D = deepcopy(data) 

93 

94 @property 

95 def data(self) -> D: 

96 """配置的原始数据*快照*""" 

97 return deepcopy(self._data) 

98 

99 @override 

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

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

102 return NotImplemented 

103 return self._data == other._data 

104 

105 __hash__ = None # type: ignore[assignment] 

106 

107 def __bool__(self) -> bool: 

108 return bool(self._data) 

109 

110 @override 

111 def __str__(self) -> str: 

112 return str(self._data) 

113 

114 @override 

115 def __repr__(self) -> str: 

116 return f"{self.__class__.__name__}({self._data!r})" 

117 

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

119 return self.from_data(self._data) 

120 

121 

122class BasicIndexedConfigData[D: Indexed[Any, Any]](BasicSingleConfigData[D], ABCIndexedConfigData[D], ABC): 

123 # noinspection GrazieInspection 

124 """ 

125 支持 ``索引`` 操作的配置数据基类 

126 

127 .. versionadded:: 0.1.5 

128 

129 .. versionchanged:: 0.2.0 

130 重命名 ``BaseSupportsIndexConfigData`` 为 ``BasicIndexedConfigData`` 

131 """ 

132 

133 def _process_path[X, Y]( 

134 self, 

135 path: ABCPath[Any], 

136 path_checker: Callable[[Any, AnyKey, ABCPath[Any], int], X], 

137 process_return: Callable[[Any], Y], 

138 ) -> X | Y: 

139 # noinspection GrazieInspection 

140 """ 

141 处理键路径的通用函数 

142 

143 :param path: 键路径 

144 :type path: ABCPath 

145 :param path_checker: 检查并处理每个路径段,返回值非None时结束操作并返回值 

146 :type path_checker: 

147 Callable[(current_data: Any, current_key: ABCKey, last_path: ABCPath, path_index: int), X] 

148 :param process_return: 处理最终结果,该函数返回值会被直接返回 

149 :type process_return: Callable[(current_data: Any), Y] 

150 

151 :return: 处理结果 

152 :rtype: X | Y 

153 

154 .. versionchanged:: 0.2.0 

155 重命名参数 ``process_check`` 为 ``path_checker`` 

156 """ # noqa: RUF002 

157 current_data = self._data 

158 

159 for key_index, current_key in enumerate(path): 

160 last_path: ABCPath[Any] = path[key_index + 1 :] 

161 

162 check_result = path_checker(current_data, current_key, last_path, key_index) 

163 if check_result is not None: 

164 return check_result 

165 

166 current_data = current_key.__get_inner_element__(current_data) 

167 

168 return process_return(current_data) 

169 

170 @override 

171 def retrieve(self, path: PathLike, *, return_raw_value: bool = False) -> Any: 

172 path = fmt_path(path) 

173 

174 def checker(current_data: Any, current_key: AnyKey, _last_path: ABCPath[Any], key_index: int) -> None: 

175 missing_protocol = current_key.__supports__(current_data) 

176 if missing_protocol: 

177 raise ConfigDataTypeError( 

178 KeyInfo(cast(ABCPath[Any], path), key_index), missing_protocol, type(current_data) 

179 ) 

180 if not current_key.__contains_inner_element__(current_data): 

181 raise RequiredPathNotFoundError(KeyInfo(cast(ABCPath[Any], path), key_index), ConfigOperate.Read) 

182 

183 def process_return[V: Any](current_data: V) -> V | ABCConfigData: 

184 if return_raw_value: 

185 return deepcopy(current_data) 

186 

187 is_sequence = isinstance(current_data, Sequence) and not isinstance(current_data, str | bytes) 

188 if isinstance(current_data, Mapping) or is_sequence: 

189 return ConfigDataFactory(current_data) # type: ignore[return-value] 

190 

191 return deepcopy(current_data) 

192 

193 return self._process_path(path, checker, process_return) 

194 

195 @override 

196 @check_read_only 

197 def modify(self, path: PathLike, value: Any, *, allow_create: bool = True) -> Self: 

198 path = fmt_path(path) 

199 

200 def checker(current_data: Any, current_key: AnyKey, last_path: ABCPath[Any], key_index: int) -> None: 

201 missing_protocol = current_key.__supports_modify__(current_data) 

202 if missing_protocol: 

203 raise ConfigDataTypeError( 

204 KeyInfo(cast(ABCPath[Any], path), key_index), missing_protocol, type(current_data) 

205 ) 

206 if not current_key.__contains_inner_element__(current_data): 

207 if not allow_create: 

208 raise RequiredPathNotFoundError(KeyInfo(cast(ABCPath[Any], path), key_index), ConfigOperate.Write) 

209 current_key.__set_inner_element__(current_data, type(self._data)()) 

210 

211 if not last_path: 

212 current_key.__set_inner_element__(current_data, value) 

213 

214 self._process_path(path, checker, lambda *_: None) 

215 return self 

216 

217 @override 

218 @check_read_only 

219 def delete(self, path: PathLike) -> Self: 

220 path = fmt_path(path) 

221 

222 def checker( 

223 current_data: Any, 

224 current_key: AnyKey, 

225 last_path: ABCPath[Any], 

226 key_index: int, 

227 ) -> Literal[True] | None: 

228 missing_protocol = current_key.__supports_modify__(current_data) 

229 if missing_protocol: 

230 raise ConfigDataTypeError( 

231 KeyInfo(cast(ABCPath[Any], path), key_index), missing_protocol, type(current_data) 

232 ) 

233 if not current_key.__contains_inner_element__(current_data): 

234 raise RequiredPathNotFoundError(KeyInfo(cast(ABCPath[Any], path), key_index), ConfigOperate.Delete) 

235 

236 if not last_path: 

237 current_key.__delete_inner_element__(current_data) 

238 return True 

239 return None # 被mypy强制要求 

240 

241 self._process_path(path, checker, lambda *_: None) 

242 return self 

243 

244 @override 

245 def unset(self, path: PathLike) -> Self: 

246 with suppress(RequiredPathNotFoundError): 

247 self.delete(path) 

248 return self 

249 

250 @override 

251 def exists(self, path: PathLike, *, ignore_wrong_type: bool = False) -> bool: 

252 path = fmt_path(path) 

253 

254 def checker(current_data: Any, current_key: AnyKey, _last_path: ABCPath[Any], key_index: int) -> bool | None: 

255 missing_protocol = current_key.__supports__(current_data) 

256 if missing_protocol: 

257 if ignore_wrong_type: 

258 return False 

259 raise ConfigDataTypeError( 

260 KeyInfo(cast(ABCPath[Any], path), key_index), missing_protocol, type(current_data) 

261 ) 

262 if not current_key.__contains_inner_element__(current_data): 

263 return False 

264 return None 

265 

266 return cast(bool, self._process_path(path, checker, lambda *_: True)) 

267 

268 @override 

269 def get[V](self, path: PathLike, default: V | None = None, *, return_raw_value: bool = False) -> V | Any: 

270 try: 

271 return self.retrieve(path, return_raw_value=return_raw_value) 

272 except RequiredPathNotFoundError: 

273 return default 

274 

275 @override 

276 def setdefault[V](self, path: PathLike, default: V | None = None, *, return_raw_value: bool = False) -> V | Any: 

277 try: 

278 return self.retrieve(path) 

279 except RequiredPathNotFoundError: 

280 self.modify(path, default) 

281 return default 

282 

283 @override 

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

285 return key in self._data # type: ignore[operator] 

286 

287 @override 

288 def __iter__(self) -> Iterator[Any]: 

289 return iter(self._data) 

290 

291 @override 

292 def __len__(self) -> int: 

293 return len(self._data) # type: ignore[arg-type] 

294 

295 @override 

296 def __getitem__(self, index: Any) -> Any: 

297 data = self._data[index] 

298 is_sequence = isinstance(data, Sequence) and not isinstance(data, str | bytes) 

299 if isinstance(data, Mapping) or is_sequence: 

300 return cast(Self, ConfigDataFactory(data)) 

301 return cast(D, deepcopy(data)) 

302 

303 @override 

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

305 self._data[index] = value # type: ignore[index] 

306 

307 @override 

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

309 del self._data[index] # type: ignore[attr-defined] 

310 

311 

312class ConfigFile[D: ABCConfigData](ABCConfigFile[D]): 

313 """配置文件类""" 

314 

315 def __init__(self, initial_config: D | Any, *, config_format: str | None = None): 

316 """ 

317 :param initial_config: 配置数据 

318 :type initial_config: D 

319 :param config_format: 配置文件的格式 

320 :type config_format: str | None 

321 

322 .. caution:: 

323 本身并未对 ``initial_config`` 参数进行深拷贝,但是 :py:class:`ConfigDataFactory` 分发的类可能会将其深拷贝 

324 

325 .. versionchanged:: 0.2.0 

326 现在会自动尝试使用 :py:class:`ConfigDataFactory` 转换 ``initial_config`` 参数 

327 

328 重命名参数 ``config_data`` 为 ``initial_config`` 

329 """ # noqa: RUF002, D205 

330 super().__init__(cast(D, ConfigDataFactory(initial_config)), config_format=config_format) 

331 

332 @override 

333 def save( 

334 self, 

335 processor_pool: ABCSLProcessorPool, 

336 namespace: str, 

337 file_name: str, 

338 config_format: str | None = None, 

339 *processor_args: Any, 

340 **processor_kwargs: Any, 

341 ) -> None: 

342 if config_format is None: 

343 config_format = self._config_format 

344 

345 if config_format not in processor_pool.SLProcessors: 

346 raise UnsupportedConfigFormatError(config_format) 

347 

348 return processor_pool.SLProcessors[config_format].save( 

349 processor_pool, self, processor_pool.root_path, namespace, file_name, *processor_args, **processor_kwargs 

350 ) 

351 

352 @classmethod 

353 @override 

354 def load( 

355 cls, 

356 processor_pool: ABCSLProcessorPool, 

357 namespace: str, 

358 file_name: str, 

359 config_format: str, 

360 *processor_args: Any, 

361 **processor_kwargs: Any, 

362 ) -> Self: 

363 if config_format not in processor_pool.SLProcessors: 

364 raise UnsupportedConfigFormatError(config_format) 

365 

366 return cast( 

367 Self, 

368 processor_pool.SLProcessors[config_format].load( 

369 processor_pool, processor_pool.root_path, namespace, file_name, *processor_args, **processor_kwargs 

370 ), 

371 ) 

372 

373 @classmethod 

374 @override 

375 def initialize( 

376 cls, 

377 processor_pool: ABCSLProcessorPool, 

378 namespace: str, 

379 file_name: str, 

380 config_format: str, 

381 *processor_args: Any, 

382 **processor_kwargs: Any, 

383 ) -> Self: 

384 if config_format not in processor_pool.SLProcessors: 

385 raise UnsupportedConfigFormatError(config_format) 

386 

387 return cast( 

388 Self, 

389 processor_pool.SLProcessors[config_format].initialize( 

390 processor_pool, processor_pool.root_path, namespace, file_name, *processor_args, **processor_kwargs 

391 ), 

392 ) 

393 

394 

395class PHelper(ABCProcessorHelper): 

396 """处理器助手类""" 

397 

398 

399class BasicConfigPool(ABCConfigPool, ABC): 

400 """ 

401 基础配置池类 

402 

403 实现了一些通用方法 

404 

405 .. versionchanged:: 0.2.0 

406 重命名 ``BaseConfigPool`` 为 ``BasicConfigPool`` 

407 """ 

408 

409 def __init__(self, root_path: str = "./.config"): 

410 """ 

411 :param root_path: 配置根路径 

412 :type root_path: str 

413 """ # noqa: D205 

414 super().__init__(root_path) 

415 self._configs: dict[str, dict[str, ABCConfigFile[Any]]] = {} 

416 self._helper = PHelper() 

417 

418 @property 

419 @override 

420 def helper(self) -> ABCProcessorHelper: 

421 return self._helper 

422 

423 # noinspection PyMethodOverriding 

424 @overload # 咱也不知道为什么mypy只有这样检查会通过而pycharm会报错 

425 def get(self, namespace: str) -> dict[str, ABCConfigFile[Any]] | None: ... 

426 

427 # noinspection PyMethodOverriding 

428 @overload 

429 def get(self, namespace: str, file_name: str) -> ABCConfigFile[Any] | None: ... 

430 

431 @overload 

432 def get( 

433 self, 

434 namespace: str, 

435 file_name: str | None = None, 

436 ) -> dict[str, ABCConfigFile[Any]] | ABCConfigFile[Any] | None: ... 

437 

438 @override 

439 def get( 

440 self, 

441 namespace: str, 

442 file_name: str | None = None, 

443 ) -> dict[str, ABCConfigFile[Any]] | ABCConfigFile[Any] | None: 

444 if namespace not in self._configs: 

445 return None 

446 result = self._configs[namespace] 

447 

448 if file_name is None: 

449 return result 

450 

451 if file_name in result: 

452 return result[file_name] 

453 

454 return None 

455 

456 @override 

457 def set(self, namespace: str, file_name: str, config: ABCConfigFile[Any]) -> Self: 

458 if namespace not in self._configs: 

459 self._configs[namespace] = {} 

460 

461 self._configs[namespace][file_name] = config 

462 return self 

463 

464 def _get_formats( 

465 self, 

466 file_name: str, 

467 config_formats: str | Iterable[str] | None, 

468 configfile_format: str | None = None, 

469 ) -> Iterable[str]: 

470 """ 

471 从给定参数计算所有可能的配置格式 

472 

473 .. attention:: 

474 返回所有可能的配置格式,不会检查配置格式是否存在! 

475 可迭代对象的产生顺序即为配置格式优先级,优先级逻辑见下表 

476 

477 :param file_name: 文件名 

478 :type file_name: str 

479 :param config_formats: 配置格式 

480 :type config_formats: str | Iterable[str] | None 

481 :param configfile_format: 

482 该配置文件对象本身配置格式属性的值 

483 可选项,一般在保存时填入 

484 用于在没手动指定配置格式且没文件后缀时使用该值进行尝试 

485 

486 .. seealso:: 

487 :py:attr:`ABCConfigFile.config_format` 

488 

489 :return: 配置格式 

490 :rtype: Iterable[str] 

491 

492 :raise UnsupportedConfigFormatError: 不支持的配置格式 

493 

494 格式计算优先级 

495 -------------- 

496 

497 1.config_formats的bool求值为真 

498 

499 2.文件名注册了对应的SL处理器 

500 

501 3.configfile_format非None 

502 

503 .. versionadded:: 0.2.0 

504 """ # noqa: RUF002 

505 result_formats = [] 

506 # 先尝试从传入的参数中获取配置文件格式 

507 if config_formats is None: 

508 config_formats = [] 

509 elif isinstance(config_formats, str): 

510 config_formats = [config_formats] 

511 else: 

512 config_formats = list(config_formats) 

513 result_formats.extend(config_formats) 

514 

515 def _check_file_name(match: str | Pattern[str]) -> bool: 

516 if isinstance(match, str): 

517 return file_name.endswith(match) 

518 return bool(match.fullmatch(file_name)) # 目前没SL处理器用得上 # pragma: no cover 

519 

520 # 再尝试从文件名匹配配置文件格式 

521 for m in self.FileNameProcessors: 

522 if _check_file_name(m): 

523 result_formats.extend(self.FileNameProcessors[m]) 

524 

525 # 最后尝试从配置文件对象本身获取配置文件格式 

526 if configfile_format is not None: 

527 result_formats.append(configfile_format) 

528 

529 if not result_formats: 

530 raise UnsupportedConfigFormatError(None) 

531 

532 return OrderedDict.fromkeys(result_formats) 

533 

534 def _try_sl_processors[R]( 

535 self, 

536 namespace: str, 

537 file_name: str, 

538 config_formats: str | Iterable[str] | None, 

539 processor: Callable[[Self, str, str, str], R], 

540 file_config_format: str | None = None, 

541 ) -> R: 

542 """ 

543 自动尝试推断ABCConfigFile所支持的config_format 

544 

545 :param namespace: 命名空间 

546 :type namespace: str 

547 :param file_name: 文件名 

548 :type file_name: str 

549 :param config_formats: 配置格式 

550 :type config_formats: str | Iterable[str] | None 

551 :param processor: 

552 处理器,参数为[配置池对象, 命名空间, 文件名, 配置格式]返回值会被直接返回, 

553 出现意料内的SL处理器无法处理需抛出FailedProcessConfigFileError以允许继续尝试别的SL处理器 

554 :type processor: Callable[[Self, str, str, str], R] 

555 :param file_config_format: 

556 该配置文件对象本身配置格式属性的值 

557 可选项,一般在保存时填入 

558 用于在没手动指定配置格式且没文件后缀时使用该值进行尝试 

559 

560 .. seealso:: 

561 :py:attr:`ABCConfigFile.config_format` 

562 

563 :return: 处理器返回值 

564 :rtype: R 

565 

566 :raise UnsupportedConfigFormatError: 不支持的配置格式 

567 :raise FailedProcessConfigFileError: 处理配置文件失败 

568 

569 .. seealso:: 

570 格式计算优先级 

571 

572 :py:meth:`_get_formats` 

573 

574 .. versionadded:: 0.1.2 

575 

576 .. versionchanged:: 0.2.0 

577 拆分格式计算到方法 :py:meth:`_get_formats` 

578 """ # noqa: RUF002 

579 

580 def callback_wrapper(cfg_fmt: str) -> R: 

581 return processor(self, namespace, file_name, cfg_fmt) 

582 

583 # 尝试从多个SL加载器中找到能正确加载的那一个 

584 errors: dict[str, FailedProcessConfigFileError[Any] | UnsupportedConfigFormatError] = {} 

585 for config_format in self._get_formats(file_name, config_formats, file_config_format): 

586 if config_format not in self.SLProcessors: 

587 errors[config_format] = UnsupportedConfigFormatError(config_format) 

588 continue 

589 try: 

590 # 能正常运行直接返回结果不再进行尝试 

591 return callback_wrapper(config_format) 

592 except FailedProcessConfigFileError as err: 

593 errors[config_format] = err 

594 

595 for error in errors.values(): 

596 if isinstance(error, UnsupportedConfigFormatError): 

597 raise error from None 

598 

599 # 如果没有一个SL加载器能正确加载则抛出异常 

600 raise FailedProcessConfigFileError(errors) 

601 

602 @override 

603 def save( 

604 self, 

605 namespace: str, 

606 file_name: str, 

607 config_formats: str | Iterable[str] | None = None, 

608 config: ABCConfigFile[Any] | None = None, 

609 *args: Any, 

610 **kwargs: Any, 

611 ) -> Self: 

612 if config is not None: 

613 self.set(namespace, file_name, config) 

614 

615 file = self._configs[namespace][file_name] 

616 

617 def processor(pool: Self, ns: str, fn: str, cf: str) -> None: 

618 file.save(pool, ns, fn, cf, *args, **kwargs) 

619 

620 self._try_sl_processors(namespace, file_name, config_formats, processor, file_config_format=file.config_format) 

621 return self 

622 

623 @override 

624 def save_all( 

625 self, *, ignore_err: bool = False 

626 ) -> dict[str, dict[str, tuple[ABCConfigFile[Any], Exception]]] | None: 

627 errors: dict[str, dict[str, tuple[ABCConfigFile[Any], Exception]]] = {} 

628 for namespace, configs in deepcopy(self._configs).items(): 

629 errors[namespace] = {} 

630 for file_name, config in configs.items(): 

631 try: 

632 self.save(namespace, file_name) 

633 except Exception as err: 

634 if not ignore_err: 

635 raise 

636 errors[namespace][file_name] = (config, err) 

637 

638 if not ignore_err: 

639 return None 

640 

641 return {k: v for k, v in errors.items() if v} 

642 

643 @override 

644 def initialize( 

645 self, 

646 namespace: str, 

647 file_name: str, 

648 *args: Any, 

649 config_formats: str | Iterable[str] | None = None, 

650 **kwargs: Any, 

651 ) -> ABCConfigFile[Any]: 

652 def processor(pool: Self, ns: str, fn: str, cf: str) -> ABCConfigFile[Any]: 

653 config_file_cls: type[ABCConfigFile[Any]] = self.SLProcessors[cf].supported_file_classes[0] 

654 result = config_file_cls.initialize(pool, ns, fn, cf, *args, **kwargs) 

655 

656 pool.set(namespace, file_name, result) 

657 return result 

658 

659 return self._try_sl_processors(namespace, file_name, config_formats, processor) 

660 

661 @override 

662 def load( 

663 self, 

664 namespace: str, 

665 file_name: str, 

666 *args: Any, 

667 config_formats: str | Iterable[str] | None = None, 

668 allow_initialize: bool = False, 

669 **kwargs: Any, 

670 ) -> ABCConfigFile[Any]: 

671 """ 

672 加载配置到指定命名空间并返回 

673 

674 :param namespace: 命名空间 

675 :type namespace: str 

676 :param file_name: 文件名 

677 :type file_name: str 

678 :param config_formats: 配置格式 

679 :type config_formats: str | Iterable[str] | None 

680 :param allow_initialize: 是否允许初始化配置文件 

681 :type allow_initialize: bool 

682 

683 :return: 配置对象 

684 :rtype: ABCConfigFile 

685 

686 .. versionchanged:: 0.2.0 

687 现在会像 :py:meth:`save` 一样接收并传递额外参数 

688 

689 删除参数 ``config_file_cls`` 

690 

691 重命名参数 ``allow_create`` 为 ``allow_initialize`` 

692 

693 现在由 :py:meth:`ABCConfigFile.initialize` 创建新的空 :py:class:`ABCConfigFile` 对象 

694 """ 

695 cache = self.get(namespace, file_name) 

696 if cache is not None: 

697 return cache 

698 

699 def processor(pool: Self, ns: str, fn: str, cf: str) -> ABCConfigFile[Any]: 

700 config_file_cls = self.SLProcessors[cf].supported_file_classes[0] 

701 try: 

702 result = config_file_cls.load(pool, ns, fn, cf, *args, **kwargs) 

703 except FileNotFoundError: 

704 if not allow_initialize: 

705 raise 

706 result = pool.initialize(ns, fn, *args, config_formats=cf, **kwargs) 

707 

708 pool.set(namespace, file_name, result) 

709 return result 

710 

711 return self._try_sl_processors(namespace, file_name, config_formats, processor) 

712 

713 @override 

714 def remove(self, namespace: str, file_name: str | None = None) -> Self: 

715 if file_name is None: 

716 del self._configs[namespace] 

717 return self 

718 

719 del self._configs[namespace][file_name] 

720 if not self._configs[namespace]: 

721 del self._configs[namespace] 

722 return self 

723 

724 @override 

725 def discard(self, namespace: str, file_name: str | None = None) -> Self: 

726 with suppress(KeyError): 

727 self.remove(namespace, file_name) 

728 return self 

729 

730 def __getitem__(self, item: str | tuple[str, str]) -> dict[str, ABCConfigFile[Any]] | ABCConfigFile[Any]: 

731 if isinstance(item, tuple): 

732 if len(item) != 2: 

733 msg = f"item must be a tuple of length 2, got {item}" 

734 raise ValueError(msg) 

735 return deepcopy(self.configs[item[0]][item[1]]) 

736 return deepcopy(self.configs[item]) 

737 

738 def __contains__(self, item: Any) -> bool: 

739 """.. versionadded:: 0.1.2""" 

740 if isinstance(item, str): 

741 return item in self._configs 

742 if isinstance(item, Iterable): 

743 item = tuple(item) 

744 if len(item) == 1: 

745 return item[0] in self._configs 

746 if len(item) != 2: 

747 msg = f"item must be a tuple of length 2, got {item}" 

748 raise ValueError(msg) 

749 return (item[0] in self._configs) and (item[1] in self._configs[item[0]]) 

750 

751 def __len__(self) -> int: 

752 """配置文件总数""" 

753 return sum(len(v) for v in self._configs.values()) 

754 

755 @property 

756 def configs(self) -> dict[str, dict[str, ABCConfigFile[Any]]]: 

757 """配置文件字典""" 

758 return deepcopy(self._configs) 

759 

760 @override 

761 def __repr__(self) -> str: 

762 return f"{self.__class__.__name__}({self.configs!r})" 

763 

764 

765__all__ = ( 

766 "BasicConfigData", 

767 "BasicConfigPool", 

768 "BasicIndexedConfigData", 

769 "BasicSingleConfigData", 

770 "ConfigFile", 

771 "PHelper", 

772)