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

1# cython: language_level = 3 # noqa: ERA001 

2 

3 

4"""主要部分""" 

5 

6import os.path 

7import re 

8from abc import ABC 

9from abc import abstractmethod 

10from collections.abc import Callable 

11from collections.abc import Generator 

12from collections.abc import Iterable 

13from collections.abc import Mapping 

14from collections.abc import Sequence 

15from contextlib import contextmanager 

16from copy import deepcopy 

17from functools import update_wrapper 

18from typing import Any 

19from typing import ClassVar 

20from typing import Literal 

21from typing import Self 

22from typing import cast 

23from typing import override 

24 

25import wrapt 

26from mypy_extensions import KwArg 

27from mypy_extensions import VarArg 

28 

29from .abc import ABCConfigData 

30from .abc import ABCConfigFile 

31from .abc import ABCConfigPool 

32from .abc import ABCConfigSL 

33from .abc import ABCSLProcessorPool 

34from .abc import SLArgumentType 

35from .basic.core import BasicConfigPool 

36from .basic.core import ConfigFile 

37from .basic.factory import ConfigDataFactory 

38from .errors import FailedProcessConfigFileError 

39from .safe_writer import safe_open 

40from .utils import FrozenArguments 

41from .utils import Ref 

42from .validators import ComponentValidatorFactory 

43from .validators import DefaultValidatorFactory 

44from .validators import ValidatorOptions 

45from .validators import ValidatorTypes 

46from .validators import pydantic_validator 

47 

48type ValidatorFactoryType[V, D: ABCConfigData] = Callable[[V, ValidatorOptions], Callable[[Ref[D]], D]] 

49""" 

50.. versionchanged:: 0.3.0 

51 重命名 ``VALIDATOR_FACTORY`` 为 ``ValidatorFactoryType`` 

52""" 

53 

54 

55class RequiredPath[V, D: ABCConfigData]: 

56 """对需求的键进行存在检查、类型检查、填充默认值""" 

57 

58 def __init__( 

59 self, 

60 validator: V, 

61 validator_factory: ValidatorFactoryType[V, D] 

62 | ValidatorTypes 

63 | Literal["custom", "pydantic", "component"] 

64 | None = ValidatorTypes.DEFAULT, 

65 static_config: ValidatorOptions | None = None, 

66 ): 

67 """ 

68 :param validator: 数据验证器 

69 :type validator: Any 

70 :param validator_factory: 数据验证器工厂 

71 :type validator_factory: 

72 ValidatorFactoryType[V, D] 

73 | validators.ValidatorTypes | Literal["custom", "pydantic", "component"] | None 

74 :param static_config: 静态配置 

75 :type static_config: ValidatorOptions | None 

76 

77 .. tip:: 

78 提供 ``static_config`` 参数可以避免在 :py:meth:`~RequiredPath.filter` 中反复调用 ``validator_factory`` 

79 以提高性能 ( :py:meth:`~RequiredPath.filter` 未传入验证器选项参数时优化生效，如果传入了则回退到默认行为) 

80 """ # noqa: RUF002, D205 

81 if not callable(validator_factory): 

82 validator_factory = ValidatorTypes(validator_factory) 

83 if isinstance(validator_factory, ValidatorTypes): 

84 validator_factory = self.ValidatorFactories[validator_factory] 

85 

86 self._validator = deepcopy(validator) 

87 self._validator_factory: ValidatorFactoryType[V, D] = validator_factory 

88 if static_config is not None: 

89 self._static_validator: Callable[[Ref[D]], D] | None = self._validator_factory( 

90 self._validator, static_config 

91 ) 

92 else: 

93 self._static_validator = None 

94 

95 ValidatorFactories: ClassVar[dict[ValidatorTypes, ValidatorFactoryType[Any, Any]]] = { 

96 ValidatorTypes.DEFAULT: cast(ValidatorFactoryType[V, D], DefaultValidatorFactory), 

97 ValidatorTypes.CUSTOM: lambda v, cfg: (lambda ref: ref.value) if v is None else lambda ref: v(ref, cfg), 

98 ValidatorTypes.PYDANTIC: cast(ValidatorFactoryType[V, D], pydantic_validator), 

99 ValidatorTypes.COMPONENT: cast(ValidatorFactoryType[V, D], ComponentValidatorFactory), 

100 } 

101 """ 

102 验证器工厂注册表 

103 

104 .. versionchanged:: 0.2.0 

105 现在待验证的配置数据必须由 :py:class:`~config.utils.Ref` 包装后传入 

106 """ 

107 

108 def filter( 

109 self, data: D | Ref[D], *, allow_modify: bool | None = None, skip_missing: bool | None = None, **extra: Any 

110 ) -> D: 

111 """ 

112 检查过滤需求的键 

113 

114 :param data: 要过滤的原始数据 

115 :type data: D | Ref[D] 

116 :param allow_modify: 

117 是否允许值不存在时修改 ``data`` 参数对象填充默认值(即使为False仍然会在结果中填充默认值，但不会修改 ``data`` 

118 参数对象) 

119 :type allow_modify: bool | None 

120 :param skip_missing: 忽略丢失的键 

121 :type skip_missing: bool | None 

122 :param extra: 额外参数 

123 :type extra: Any 

124 

125 :return: 处理后的配置数据*快照* 

126 :rtype: D 

127 

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

129 :raise RequiredPathNotFoundError: 必要的键未找到 

130 :raise UnknownErrorDuringValidateError: 验证过程中发生未知错误 

131 

132 .. attention:: 

133 返回的配置数据是 `*快照*` 

134 

135 .. caution:: 

136 提供了任意配置参数(``allow_modify``, ``skip_missing``, ...)时,这次调用将完全舍弃 `static_config` 

137 使用当前提供的配置参数 

138 

139 这会导致调用 `validator_factory` 产生额外开销(如果你提供 `static_config` 参数是为了避免反复调用 

140 `validator_factory` 的话) 

141 

142 .. versionchanged:: 0.2.0 

143 重命名参数 ``ignore_missing`` 为 ``skip_missing`` 

144 

145 ``data`` 参数支持 :py:class:`Ref` 

146 """ # noqa: RUF002 

147 config_kwargs: dict[str, Any] = {} 

148 if allow_modify is not None: 

149 config_kwargs["allow_modify"] = allow_modify 

150 if skip_missing is not None: 

151 config_kwargs["skip_missing"] = skip_missing 

152 if extra: 

153 config_kwargs["extra"] = extra 

154 

155 if not isinstance(data, Ref): 

156 data = Ref(data) 

157 

158 if (self._static_validator is None) or config_kwargs: 

159 config = ValidatorOptions(**config_kwargs) 

160 validator: Callable[[Ref[D]], D] = self._validator_factory(self._validator, config) 

161 else: 

162 validator = self._static_validator 

163 

164 return validator(data) 

165 

166 

167class ConfigRequirementDecorator: 

168 """ 

169 配置获取器，可作装饰器使用 

170 

171 .. versionchanged:: 0.2.0 

172 重命名 ``RequireConfigDecorator`` 为 ``ConfigRequirementDecorator`` 

173 

174 .. versionchanged:: 0.3.0 

175 修正配置加载逻辑，现在会在每一次获取配置数据时尝试加载而不是仅在初始化时尝试加载 

176 """ # noqa: RUF002 

177 

178 def __init__[D: ABCConfigData]( 

179 self, 

180 config_pool: ABCConfigPool, 

181 namespace: str, 

182 file_name: str, 

183 required: RequiredPath[Any, D], 

184 *, 

185 validate_only: bool = True, 

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

187 allow_initialize: bool = True, 

188 config_cacher: Callable[[Callable[..., D], VarArg(), KwArg()], D] | None = None, 

189 filter_kwargs: dict[str, Any] | None = None, 

190 ): 

191 # noinspection GrazieInspection 

192 """ 

193 :param config_pool: 所在的配置池 

194 :type config_pool: ConfigPool 

195 :param namespace: 详见 :py:meth:`ConfigPool.load` 

196 :param file_name: 详见 :py:meth:`ConfigPool.load` 

197 :param required: 需求的键 

198 :type required: RequiredPath[Any, D] 

199 :param validate_only: 

200 如果为 :py:const:`True` 则返回配置池中完整的配置数据（按引用传递），否则返回验证后的数据（按值传递）， 

201 此选项有助于避免意外修改配置池中的数据 

202 :type validate_only: bool 

203 :param config_formats: 详见 :py:meth:`ConfigPool.load` 

204 :param allow_initialize: 详见 :py:meth:`ConfigPool.load` 

205 :param config_cacher: 缓存配置的装饰器，默认为None，即不缓存 

206 :type config_cacher: Callable[[Callable[..., D], VarArg(), KwArg()], D] | None 

207 :param filter_kwargs: :py:meth:`RequiredPath.filter` 要绑定的默认参数，这会导致 ``static_config`` 失效 

208 :type filter_kwargs: dict[str, Any] | None 

209 

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

211 

212 .. versionchanged:: 0.2.0 

213 重命名参数 ``cache_config`` 为 ``config_cacher`` 

214 

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

216 

217 .. versionadded:: 0.3.0 

218 添加参数 ``validate_only`` 

219 """ # noqa: RUF002, D205 

220 if filter_kwargs is None: 

221 filter_kwargs = {} 

222 

223 self._config_loader: Callable[[], ABCConfigFile[D]] = lambda: config_pool.load( 

224 namespace, file_name, config_formats=config_formats, allow_initialize=allow_initialize 

225 ) 

226 self._required = required 

227 self._validate_only = validate_only 

228 self._filter_kwargs = filter_kwargs 

229 # noinspection PyInvalidCast 

230 self._config_cacher: Callable[[Callable[..., D], VarArg(), KwArg()], D] = ( 

231 cast( 

232 Callable[[Callable[..., D], VarArg(), KwArg()], D], lambda func, *args, **kwargs: func(*args, **kwargs) 

233 ) 

234 if config_cacher is None 

235 else config_cacher 

236 ) 

237 

238 def check(self, *, ignore_cache: bool = False, **filter_kwargs: Any) -> Any: 

239 """ 

240 手动检查配置 

241 

242 :param ignore_cache: 是否忽略缓存 

243 :type ignore_cache: bool 

244 :param filter_kwargs: RequiredConfig.filter的参数 

245 

246 :return: 得到的配置数据 

247 :rtype: Any 

248 

249 """ 

250 kwargs = self._filter_kwargs | filter_kwargs 

251 

252 if ignore_cache: 

253 config_ref = Ref((config_file := self._config_loader()).config) 

254 result = self._required.filter(config_ref, **kwargs) 

255 config_file._config = config_ref.value # noqa: SLF001 

256 

257 return result 

258 return self._wrapped_filter(**kwargs) 

259 

260 def __call__(self, func: Callable[[ABCConfigData, Any], Any]) -> Callable[..., Any]: 

261 """ 

262 通过装饰器提供配置数据注入，配置数据将会注入到 ``self`` (如果为方法而不是函数)后的第一个参数 

263 

264 :param func: 需要装饰的函数 

265 :type func: Callable[[ABCConfigData, Any], Any] 

266 

267 :return: 装饰后的函数 

268 :rtype: Callable[..., Any] 

269 """ # noqa: RUF002 

270 

271 @wrapt.decorator 

272 def wrapper( 

273 wrapped: Callable[..., Any], 

274 _instance: object | None, 

275 args: tuple[Any, ...], 

276 kwargs: dict[str, Any], 

277 ) -> Any: 

278 config_data = self._wrapped_filter(**self._filter_kwargs) 

279 

280 return wrapped(config_data, *args, **kwargs) 

281 

282 return cast(Callable[..., Any], update_wrapper(wrapper(func), func)) 

283 

284 def _wrapped_filter(self, **kwargs: Any) -> ABCConfigData: 

285 config_ref = Ref((config_file := self._config_loader()).config) 

286 

287 result = self._config_cacher(self._required.filter, config_ref, **kwargs) 

288 config_file._config = config_ref.value # noqa: SLF001 

289 if self._validate_only: 

290 return config_file.config 

291 return deepcopy(result) if config_file.config is result else result 

292 

293 

294class ConfigPool(BasicConfigPool): 

295 """配置池""" 

296 

297 def require( 

298 self, 

299 namespace: str, 

300 file_name: str, 

301 validator: Any, 

302 validator_factory: Any = ValidatorTypes.DEFAULT, 

303 static_config: Any | None = None, 

304 **kwargs: Any, 

305 ) -> ConfigRequirementDecorator: 

306 # noinspection GrazieInspection 

307 """ 

308 获取配置 

309 

310 :param namespace: 命名空间 

311 :type namespace: str 

312 :param file_name: 文件名 

313 :type file_name: str 

314 :param validator: 详见 :py:class:`RequiredPath` 

315 :param validator_factory: 详见 :py:class:`RequiredPath` 

316 :param static_config: 详见 :py:class:`RequiredPath` 

317 

318 :param kwargs: 详见 :py:class:`ConfigRequirementDecorator` 

319 

320 :return: 详见 :py:class:`ConfigRequirementDecorator` 

321 :rtype: :py:class:`ConfigRequirementDecorator` 

322 

323 .. versionchanged:: 0.2.0 

324 删除声明于 ``ABCConfigPool`` 

325 """ 

326 return ConfigRequirementDecorator( 

327 self, namespace, file_name, RequiredPath(validator, validator_factory, static_config), **kwargs 

328 ) 

329 

330 

331DefaultConfigPool = ConfigPool() 

332""" 

333默认配置池 

334""" 

335requireConfig = DefaultConfigPool.require # noqa: N816 

336""" 

337:py:data:`DefaultConfigPool` . :py:meth:`~ConfigPool.require` 

338""" 

339saveAll = DefaultConfigPool.save_all # noqa: N816 

340""" 

341:py:data:`DefaultConfigPool` . :py:meth:`~ConfigPool.save_all` 

342""" 

343get = DefaultConfigPool.get 

344""" 

345:py:data:`DefaultConfigPool` . :py:meth:`~ConfigPool.get` 

346""" 

347set_ = DefaultConfigPool.set 

348""" 

349:py:data:`DefaultConfigPool` . :py:meth:`~ConfigPool.set` 

350""" 

351save = DefaultConfigPool.save 

352""" 

353:py:data:`DefaultConfigPool` . :py:meth:`~ConfigPool.save` 

354""" 

355load = DefaultConfigPool.load 

356""" 

357:py:data:`DefaultConfigPool` . :py:meth:`~ConfigPool.load` 

358""" 

359 

360 

361class BasicConfigSL(ABCConfigSL, ABC): 

362 """ 

363 基础配置SL管理器 提供了一些实用功能 

364 

365 .. versionchanged:: 0.2.0 

366 重命名 ``BaseConfigSL`` 为 ``BasicConfigSL`` 

367 """ 

368 

369 @override 

370 def register_to(self, config_pool: ABCSLProcessorPool | None = None) -> Self: 

371 """ 

372 注册到配置池中 

373 

374 :param config_pool: 配置池 

375 :type config_pool: ABCSLProcessorPool | None 

376 

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

378 :rtype: Self 

379 

380 .. versionchanged:: 0.3.0 

381 返回当前实例便于链式调用 

382 """ 

383 if config_pool is None: 

384 config_pool = DefaultConfigPool 

385 

386 return super().register_to(config_pool) 

387 

388 @override 

389 def initialize( 

390 self, 

391 processor_pool: ABCSLProcessorPool, 

392 root_path: str, 

393 namespace: str, 

394 file_name: str, 

395 *args: Any, 

396 **kwargs: Any, 

397 ) -> ABCConfigFile[Any]: 

398 return ConfigFile(ConfigDataFactory(), config_format=self.reg_name) 

399 

400 

401@contextmanager 

402def raises(excs: type[Exception] | tuple[type[Exception], ...] = Exception) -> Generator[None, Any, None]: 

403 """ 

404 包装意料内的异常 

405 

406 提供给子类的便捷方法 

407 

408 :param excs: 意料内的异常 

409 :type excs: type[Exception] | tuple[type[Exception], ...] 

410 

411 :raise FailedProcessConfigFileError: 当触发了对应的异常时 

412 

413 .. versionadded:: 0.1.4 

414 

415 .. versionchanged:: 0.2.0 

416 提取为函数 

417 """ 

418 try: 

419 yield 

420 except excs as err: 

421 raise FailedProcessConfigFileError(err) from None 

422 

423 

424class BasicLocalFileConfigSL(BasicConfigSL, ABC): 

425 """ 

426 基础本地配置文件SL处理器 

427 

428 .. versionchanged:: 0.2.0 

429 重命名从 ``BaseLocalFileConfigSL`` 为 ``BasicLocalFileConfigSL`` 

430 """ 

431 

432 _s_open_kwargs: dict[str, Any] = {"mode": "w", "encoding": "utf-8"} # noqa: RUF012 

433 _l_open_kwargs: dict[str, Any] = {"mode": "r", "encoding": "utf-8"} # noqa: RUF012 

434 

435 def __init__( 

436 self, 

437 s_arg: SLArgumentType = None, 

438 l_arg: SLArgumentType = None, 

439 *, 

440 reg_alias: str | None = None, 

441 create_dir: bool = True, 

442 ): 

443 # noinspection GrazieInspection 

444 """ 

445 :param s_arg: 保存器默认参数 

446 :type s_arg: SLArgumentType 

447 :param l_arg: 加载器默认参数 

448 :type l_arg: SLArgumentType 

449 :param reg_alias: sl处理器注册别名 

450 :type reg_alias: Optional[str] 

451 :param create_dir: 是否允许创建目录 

452 :type create_dir: bool 

453 

454 .. versionchanged:: 0.2.0 

455 将 ``保存加载器参数`` 相关从 :py:class:`BasicConfigSL` 移动到此类 

456 """ # noqa: D205 

457 

458 def _build_arg(value: SLArgumentType) -> FrozenArguments: 

459 sl_args: tuple[()] | tuple[Sequence[Any] | None, Mapping[str, Any] | None] 

460 if value is None: 

461 sl_args = () 

462 elif isinstance(value, Sequence): 

463 sl_args = value, None 

464 elif isinstance(value, Mapping): 

465 sl_args = None, value 

466 else: 

467 msg = f"Invalid argument type, must be '{SLArgumentType}'" 

468 raise TypeError(msg) 

469 return FrozenArguments(*sl_args) 

470 

471 self._saver_args: FrozenArguments = _build_arg(s_arg) 

472 self._loader_args: FrozenArguments = _build_arg(l_arg) 

473 

474 super().__init__(reg_alias=reg_alias) 

475 

476 self.create_dir = create_dir 

477 

478 @property 

479 def saver_args(self) -> FrozenArguments: 

480 """保存器默认参数""" 

481 return self._saver_args 

482 

483 @property 

484 def loader_args(self) -> FrozenArguments: 

485 """加载器默认参数""" 

486 return self._loader_args 

487 

488 raises = staticmethod(raises) 

489 

490 @override 

491 def save( 

492 self, 

493 processor_pool: ABCSLProcessorPool, 

494 config_file: ABCConfigFile[Any], 

495 root_path: str, 

496 namespace: str, 

497 file_name: str, 

498 *args: Any, 

499 **kwargs: Any, 

500 ) -> None: 

501 """ 

502 保存处理器 (原子操作 多线/进程安全) 

503 

504 :param processor_pool: 配置池 

505 :type processor_pool: ABCSLProcessorPool 

506 :param config_file: 待保存配置 

507 :type config_file: ABCConfigFile 

508 :param root_path: 保存的根目录 

509 :type root_path: str 

510 :param namespace: 配置的命名空间 

511 :type namespace: str 

512 :param file_name: 配置文件名 

513 :type file_name: str 

514 

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

516 

517 .. versionchanged:: 0.2.0 

518 现在操作是原子的(操作过程发生异常会回滚操作) 

519 

520 现在操作是理论上是多线/进程安全的 

521 

522 添加参数 ``processor_pool`` 

523 """ 

524 merged_arguments: FrozenArguments = self._saver_args | (args, kwargs) 

525 

526 file_path = processor_pool.helper.calc_path(root_path, namespace, file_name) 

527 if self.create_dir: 

528 os.makedirs(os.path.dirname(file_path), exist_ok=True) 

529 with safe_open(file_path, **self._s_open_kwargs) as f: 

530 self.save_file(config_file, f, *merged_arguments.args, **merged_arguments.kwargs) 

531 

532 @override 

533 def load( 

534 self, 

535 processor_pool: ABCSLProcessorPool, 

536 root_path: str, 

537 namespace: str, 

538 file_name: str, 

539 *args: Any, 

540 **kwargs: Any, 

541 ) -> ABCConfigFile[Any]: 

542 """ 

543 加载处理器 

544 

545 :param processor_pool: 配置池 

546 :type processor_pool: ABCSLProcessorPool 

547 :param root_path: 保存的根目录 

548 :type root_path: str 

549 :param namespace: 配置的命名空间 

550 :type namespace: str 

551 :param file_name: 配置文件名 

552 :type file_name: str 

553 

554 :return: 配置对象 

555 :rtype: ABCConfigFile 

556 

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

558 

559 .. versionchanged:: 0.2.0 

560 现在操作是原子的(操作过程发生异常会回滚操作) 

561 

562 现在操作是理论上是多线/进程安全的 

563 

564 删除参数 ``config_file_cls`` 

565 

566 添加参数 ``processor_pool`` 

567 """ 

568 merged_arguments: FrozenArguments = self._loader_args | (args, kwargs) 

569 

570 file_path = processor_pool.helper.calc_path(root_path, namespace, file_name) 

571 if self.create_dir: 

572 os.makedirs(os.path.dirname(file_path), exist_ok=True) 

573 with safe_open(file_path, **self._l_open_kwargs) as f: 

574 return self.load_file(f, *merged_arguments.args, **merged_arguments.kwargs) 

575 

576 @abstractmethod 

577 def save_file( 

578 self, 

579 config_file: ABCConfigFile[Any], 

580 target_file: Any, 

581 *merged_args: Any, 

582 **merged_kwargs: Any, 

583 ) -> None: 

584 """ 

585 将配置保存到文件 

586 

587 :param config_file: 配置文件 

588 :type config_file: ABCConfigFile 

589 :param target_file: 目标文件对象 

590 :type target_file: Any 

591 :param merged_args: 合并后的位置参数 

592 :param merged_kwargs: 合并后的关键字参数 

593 

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

595 

596 .. versionchanged:: 0.2.0 

597 更改 ``target_file`` 参数类型为 ``Any`` 

598 """ 

599 

600 @abstractmethod 

601 def load_file( 

602 self, 

603 source_file: Any, 

604 *merged_args: Any, 

605 **merged_kwargs: Any, 

606 ) -> ABCConfigFile[Any]: 

607 """ 

608 从文件加载配置 

609 

610 :param source_file: 源文件对象 

611 :type source_file: Any 

612 :param merged_args: 合并后的位置参数 

613 :param merged_kwargs: 合并后的关键字参数 

614 

615 :return: 本地配置文件对象 

616 :rtype: ABCConfigFile 

617 

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

619 

620 .. versionchanged:: 0.2.0 

621 删除参数 ``config_file_cls`` 

622 

623 更改 ``source_file`` 参数类型为 ``Any`` 

624 """ 

625 

626 @override 

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

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

629 return NotImplemented 

630 saver_args_eq = self._saver_args == other._saver_args 

631 loader_args_eq = self._loader_args == other._loader_args 

632 

633 return all((super().__eq__(other), saver_args_eq, loader_args_eq)) 

634 

635 @override 

636 def __hash__(self) -> int: 

637 return hash((super().__hash__(), self._saver_args, self._loader_args)) 

638 

639 

640class BasicChainConfigSL(BasicConfigSL, ABC): 

641 """ 

642 基础连锁配置文件SL处理器 

643 

644 .. caution:: 

645 会临时在配置文件池中添加文件以传递SL操作 

646 

647 .. versionadded:: 0.2.0 

648 """ 

649 

650 def __init__(self, *, reg_alias: str | None = None, create_dir: bool = True): 

651 """ 

652 :param reg_alias: 处理器别名 

653 :type reg_alias: Optional[str] 

654 :param create_dir: 是否创建目录 

655 :type create_dir: bool 

656 """ # noqa: D205 

657 super().__init__(reg_alias=reg_alias) 

658 

659 self.create_dir = create_dir 

660 self._cleanup_registry: bool = True 

661 """ 

662 自动清理为了传递SL处理所加入配置池的配置文件 

663 """ 

664 

665 raises = staticmethod(raises) 

666 

667 def namespace_formatter(self, namespace: str, file_name: str) -> str: # noqa: ARG002 

668 """ 

669 格式化命名空间以传递给其他SL处理器 

670 

671 :param namespace: 配置的命名空间 

672 :type namespace: str 

673 :param file_name: 配置文件名 

674 :type file_name: str 

675 

676 :return: 格式化后的命名空间 

677 :rtype: str 

678 """ 

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

680 

681 def filename_formatter(self, file_name: str) -> str: 

682 # noinspection SpellCheckingInspection 

683 """ 

684 格式化文件名以传递给其他SL处理器 

685 

686 :param file_name: 配置文件名 

687 :type file_name: str 

688 

689 :return: 格式化后的文件名 

690 :rtype: str 

691 

692 默认实现: 

693 - 遍历 :py:attr:`BasicCompressedConfigSL` 

694 - 如果为 ``str`` 且 ``file_name.endswith`` 成立则返回移除后缀后的结果 

695 - 如果为 ``re.Pattern`` 且 ``Pattern.fullmatch(file_name)`` 成立则返回 ``Pattern.sub(file_name, "")`` 

696 - 直接返回 

697 """ 

698 for match in self.supported_file_patterns: 

699 if isinstance(match, str) and file_name.endswith(match): 

700 return file_name[: -len(match)] 

701 if isinstance(match, re.Pattern) and match.fullmatch(file_name): # 目前没SL处理器用得上 # pragma: no cover 

702 return match.sub(file_name, "") 

703 return file_name # 不好测试 # pragma: no cover 

704 

705 @override 

706 def save( 

707 self, 

708 processor_pool: ABCSLProcessorPool, 

709 config_file: ABCConfigFile[Any], 

710 root_path: str, 

711 namespace: str, 

712 file_name: str, 

713 *args: Any, 

714 **kwargs: Any, 

715 ) -> None: 

716 config_pool = cast(ABCConfigPool, processor_pool) 

717 file_path = config_pool.helper.calc_path(root_path, namespace, file_name) 

718 if self.create_dir: 

719 os.makedirs(os.path.dirname(file_path), exist_ok=True) 

720 

721 formatted_filename = self.filename_formatter(file_name) 

722 formatted_namespace = self.namespace_formatter(namespace, file_name) 

723 

724 self.save_file(config_pool, config_file, formatted_namespace, formatted_filename, *args, **kwargs) 

725 self.after_save(config_pool, config_file, file_path, root_path, formatted_namespace, formatted_filename) 

726 

727 @override 

728 def load( 

729 self, 

730 processor_pool: ABCSLProcessorPool, 

731 root_path: str, 

732 namespace: str, 

733 file_name: str, 

734 *args: Any, 

735 **kwargs: Any, 

736 ) -> ABCConfigFile[Any]: 

737 config_pool = cast(ABCConfigPool, processor_pool) 

738 file_path = config_pool.helper.calc_path(root_path, namespace, file_name) 

739 if self.create_dir: 

740 os.makedirs(os.path.dirname(file_path), exist_ok=True) 

741 

742 formatted_filename = self.filename_formatter(file_name) 

743 formatted_namespace = self.namespace_formatter(namespace, file_name) 

744 

745 self.before_load(config_pool, file_path, root_path, formatted_namespace, formatted_filename) 

746 return self.load_file(config_pool, formatted_namespace, formatted_filename, *args, **kwargs) 

747 

748 @override 

749 def initialize( 

750 self, 

751 processor_pool: ABCSLProcessorPool, 

752 root_path: str, 

753 namespace: str, 

754 file_name: str, 

755 *args: Any, 

756 **kwargs: Any, 

757 ) -> ABCConfigFile[Any]: 

758 config_pool = cast(ABCConfigPool, processor_pool) 

759 

760 formatted_namespace = self.namespace_formatter(namespace, file_name) 

761 formatted_filename = self.filename_formatter(file_name) 

762 

763 return config_pool.initialize(formatted_namespace, formatted_filename, *args, **kwargs) 

764 

765 def save_file( 

766 self, 

767 config_pool: ABCConfigPool, 

768 config_file: ABCConfigFile[Any], 

769 namespace: str, 

770 file_name: str, 

771 *args: Any, 

772 **kwargs: Any, 

773 ) -> None: 

774 """ 

775 保存指定命名空间的配置 

776 

777 :param config_pool: 配置池 

778 :type config_pool: ABCConfigPool 

779 :param config_file: 配置文件 

780 :type config_file: ABCConfigFile 

781 :param namespace: 命名空间 

782 :type namespace: str 

783 :param file_name: 文件名 

784 :type file_name: str 

785 """ 

786 config_pool.save(namespace, file_name, *args, config=config_file, **kwargs) # type: ignore[misc] 

787 if self._cleanup_registry: 

788 config_pool.discard(namespace, file_name) 

789 

790 def load_file( 

791 self, 

792 config_pool: ABCConfigPool, 

793 namespace: str, 

794 file_name: str, 

795 *args: Any, 

796 **kwargs: Any, 

797 ) -> ABCConfigFile[Any]: 

798 """ 

799 加载指定命名空间的配置 

800 

801 :param config_pool: 配置池 

802 :type config_pool: ABCConfigPool 

803 :param namespace: 命名空间 

804 :type namespace: str 

805 :param file_name: 文件名 

806 :type file_name: str 

807 

808 :return: 配置文件 

809 :rtype: ABCConfigFile[Any] 

810 

811 .. caution:: 

812 传递SL处理前没有清理已经缓存在配置池里的配置文件，返回的可能不是最新数据 

813 """ # noqa: RUF002 

814 cfg_file = config_pool.load(namespace, file_name, *args, **kwargs) 

815 if self._cleanup_registry: 

816 config_pool.discard(namespace, file_name) 

817 return cfg_file 

818 

819 def before_load( 

820 self, 

821 config_pool: ABCConfigPool, 

822 file_path: str, 

823 root_path: str, 

824 namespace: str, 

825 file_name: str, 

826 ) -> None: 

827 """ 

828 加载前处理 

829 

830 :param config_pool: 配置池 

831 :type config_pool: ABCConfigPool 

832 :param file_path: 文件路径 

833 :type file_path: str 

834 :param root_path: 根路径 

835 :type root_path: str 

836 :param namespace: 命名空间 

837 :type namespace: str 

838 :param file_name: 文件名 

839 :type file_name: str 

840 """ 

841 

842 def after_save( 

843 self, 

844 config_pool: ABCConfigPool, 

845 config_file: ABCConfigFile[Any], 

846 file_path: str, 

847 root_path: str, 

848 namespace: str, 

849 file_name: str, 

850 ) -> None: 

851 """ 

852 保存后处理 

853 

854 :param config_pool: 配置池 

855 :type config_pool: ABCConfigPool 

856 :param config_file: 配置文件 

857 :type config_file: ABCConfigFile[Any] 

858 :param file_path: 文件路径 

859 :type file_path: str 

860 :param root_path: 根路径 

861 :type root_path: str 

862 :param namespace: 命名空间 

863 :type namespace: str 

864 :param file_name: 文件名 

865 :type file_name: str 

866 """ 

867 

868 

869class BasicCachedConfigSL(BasicChainConfigSL, ABC): 

870 """基础缓存配置处理器""" 

871 

872 @property 

873 def namespace_suffix(self) -> str: 

874 """命名空间后缀""" 

875 return "$temporary~" 

876 

877 @override 

878 def namespace_formatter(self, namespace: str, file_name: str) -> str: 

879 return os.path.normpath(os.path.join(namespace, self.namespace_suffix, file_name)) 

880 

881 

882class BasicCompressedConfigSL(BasicCachedConfigSL, ABC): 

883 """ 

884 基础压缩配置文件SL处理器 

885 

886 .. versionadded:: 0.2.0 

887 """ 

888 

889 @property 

890 @override 

891 def namespace_suffix(self) -> str: 

892 return super().namespace_suffix 

893 

894 @override 

895 def after_save( 

896 self, 

897 config_pool: ABCConfigPool, 

898 config_file: ABCConfigFile[Any], 

899 file_path: str, 

900 root_path: str, 

901 namespace: str, 

902 file_name: str, 

903 ) -> None: 

904 extract_dir = config_pool.helper.calc_path(root_path, namespace) 

905 self.compress_file(file_path, extract_dir) 

906 

907 @override 

908 def before_load( 

909 self, 

910 config_pool: ABCConfigPool, 

911 file_path: str, 

912 root_path: str, 

913 namespace: str, 

914 file_name: str, 

915 ) -> None: 

916 extract_dir = config_pool.helper.calc_path(root_path, namespace) 

917 self.extract_file(file_path, extract_dir) 

918 

919 @abstractmethod 

920 def compress_file(self, file_path: str, extract_dir: str) -> None: 

921 """ 

922 压缩文件 

923 

924 :param file_path: 压缩文件路径 

925 :type file_path: str 

926 

927 :param extract_dir: 解压目录 

928 :type extract_dir: str 

929 """ 

930 

931 @abstractmethod 

932 def extract_file(self, file_path: str, extract_dir: str) -> None: 

933 """ 

934 解压文件 

935 

936 :param file_path: 压缩文件路径 

937 :type file_path: str 

938 

939 :param extract_dir: 解压目录