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

274 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 dataclasses 

7import re 

8import types 

9import warnings 

10from collections import OrderedDict 

11from collections.abc import Callable 

12from collections.abc import Iterable 

13from collections.abc import Mapping 

14from contextlib import suppress 

15from copy import deepcopy 

16from dataclasses import dataclass 

17from enum import Enum 

18from typing import Any 

19from typing import NamedTuple 

20from typing import Never 

21from typing import TypeAliasType 

22from typing import cast 

23from typing import overload 

24from typing import override 

25 

26from pydantic import BaseModel 

27from pydantic import ConfigDict 

28from pydantic import ValidationError 

29from pydantic import create_model 

30 

31# noinspection PyProtectedMember 

32from pydantic.fields import FieldInfo 

33from pydantic_core import core_schema 

34 

35from .abc import ABCIndexedConfigData 

36from .abc import ABCPath 

37from .basic.component import ComponentConfigData 

38from .basic.mapping import MappingConfigData 

39from .basic.object import NoneConfigData 

40from .errors import ComponentMemberMismatchError 

41from .errors import ConfigDataTypeError 

42from .errors import ConfigOperate 

43from .errors import KeyInfo 

44from .errors import RequiredPathNotFoundError 

45from .errors import UnknownErrorDuringValidateError 

46from .path import Path 

47from .utils import Ref 

48from .utils import Unset 

49from .utils import UnsetType 

50from .utils import singleton 

51 

52 

53class ValidatorTypes(Enum): 

54 """验证器类型""" 

55 

56 DEFAULT = None 

57 CUSTOM = "custom" 

58 """ 

59 .. versionchanged:: 0.2.0 

60 重命名 ``IGNORE`` 为 ``NO_VALIDATION`` 

61 

62 .. versionchanged:: 0.3.0 

63 重命名 ``NO_VALIDATION`` 为 ``CUSTOM`` 

64 """ 

65 PYDANTIC = "pydantic" 

66 COMPONENT = "component" 

67 """ 

68 .. versionadded:: 0.2.0 

69 """ 

70 

71 

72@dataclass(kw_only=True) 

73class ValidatorOptions: 

74 # noinspection GrazieInspection 

75 """ 

76 验证器选项 

77 

78 .. versionchanged:: 0.3.0 

79 重命名 ``ValidatorFactoryConfig`` 为 ``ValidatorOptions`` 

80 """ 

81 

82 allow_modify: bool = True 

83 """ 

84 是否允许在填充默认值时同步填充源数据 

85 

86 .. versionchanged:: 0.1.2 

87 重命名 ``allow_create`` 为 ``allow_modify`` 

88 

89 .. versionchanged:: 0.2.0 

90 现在默认为 :py:const:`True` 

91 """ 

92 skip_missing: bool = False 

93 """ 

94 是否忽略不存在的路径 

95 

96 .. versionchanged:: 0.2.0 

97 重命名 ``ignore_missing`` 为 ``skip_missing`` 

98 """ 

99 

100 extra: dict[str, Any] = dataclasses.field(default_factory=dict) 

101 

102 

103type MCD = MappingConfigData[Any] 

104type ICD = ABCIndexedConfigData[Any] 

105 

106 

107# noinspection PyNewStyleGenericSyntax 

108def _remove_skip_missing[D: dict[str, Any] | list[Any]](data: D) -> D: 

109 """ 

110 递归删除值为 :py:const:`SkipMissing` 的项 

111 

112 :param data: 配置数据 

113 :type data: dict | list 

114 

115 .. versionadded:: 0.3.0 

116 """ 

117 if isinstance(data, dict): 

118 return type(data)((k, _remove_skip_missing(v)) for k, v in data.items() if v is not SkipMissing) 

119 if isinstance(data, list): 

120 return type(data)(_remove_skip_missing(item) for item in data if item is not SkipMissing) 

121 return data 

122 

123 

124def _process_pydantic_exceptions(err: ValidationError) -> Exception: 

125 """ 

126 转换包装 pydantic 的异常 

127 

128 :param err: pydantic 的异常 

129 :type err: ValidationError 

130 

131 :return: 转换后的异常 

132 :rtype: Exception 

133 """ 

134 e = err.errors()[0] 

135 

136 path = Path.from_locate(e["loc"]) 

137 

138 kwargs: dict[str, Any] = {"key_info": KeyInfo(path, -1)} 

139 

140 class ErrInfo(NamedTuple): 

141 err_type: type[Exception] | Callable[..., Exception] 

142 kwargs: dict[str, Any] 

143 

144 err_input = e["input"] 

145 err_msg = e["msg"] 

146 

147 types_kwarg: dict[str, Callable[[], ErrInfo]] = { 

148 "missing": lambda: ErrInfo(RequiredPathNotFoundError, {"operate": ConfigOperate.Read}), 

149 "model_type": lambda: ErrInfo( 

150 ConfigDataTypeError, 

151 { 

152 "required_type": ( 

153 Never if (match := re.match(r"Input should be (.*)", err_msg)) is None else match.group(1) 

154 ), 

155 "current_type": type(err_input), 

156 }, 

157 ), 

158 "int_type": lambda: ErrInfo(ConfigDataTypeError, {"required_type": int, "current_type": type(err_input)}), 

159 "int_parsing": lambda: ErrInfo(ConfigDataTypeError, {"required_type": int, "current_type": type(err_input)}), 

160 "string_type": lambda: ErrInfo(ConfigDataTypeError, {"required_type": str, "current_type": type(err_input)}), 

161 "dict_type": lambda: ErrInfo(ConfigDataTypeError, {"required_type": dict, "current_type": type(err_input)}), 

162 "literal_error": lambda: ErrInfo(RequiredPathNotFoundError, {"operate": ConfigOperate.Write}), 

163 } 

164 

165 err_type_processor = types_kwarg.get(e["type"]) 

166 if err_type_processor is None: # pragma: no cover 

167 raise UnknownErrorDuringValidateError(**kwargs, error=e) from err 

168 err_info = err_type_processor() 

169 return err_info.err_type(**(kwargs | err_info.kwargs)) 

170 

171 

172@singleton 

173class SkipMissingType: 

174 """ 

175 用于表明值可以缺失特殊值 

176 

177 .. versionchanged:: 0.2.0 

178 重命名 ``IgnoreMissingType`` 为 ``SkipMissingType`` 

179 """ 

180 

181 @override 

182 def __str__(self) -> str: 

183 return "<SkipMissing>" 

184 

185 @staticmethod 

186 def __get_pydantic_core_schema__(*_: Any) -> core_schema.ChainSchema: 

187 # 构造一个永远无法匹配的schema使 `SkipMissing | int` 可以正常工作 即被pydantic视为 `int` 

188 return core_schema.chain_schema([core_schema.none_schema(), core_schema.is_subclass_schema(type)]) 

189 

190 

191SkipMissing = SkipMissingType() 

192type AnyTypeHint = type | types.UnionType | types.EllipsisType | types.GenericAlias | TypeAliasType 

193 

194 

195@dataclass(init=False) 

196class FieldDefinition[T: AnyTypeHint]: 

197 """ 

198 字段定义,包含类型注解和默认值 

199 

200 .. versionchanged:: 0.1.4 

201 新增字段 ``allow_recursive`` 

202 

203 .. versionchanged:: 0.3.0 

204 新增支持 :py:class:`TypeAliasType` 

205 """ # noqa: RUF002 

206 

207 @overload 

208 def __init__(self, annotation: T, default: Any, *, allow_recursive: bool = True): ... 

209 

210 @overload 

211 def __init__(self, annotation: T, *, default_factory: Callable[[], Any], allow_recursive: bool = True): ... 

212 

213 def __init__( 

214 self, 

215 annotation: T, 

216 default: Any = Unset, 

217 *, 

218 default_factory: Callable[[], Any] | UnsetType = Unset, 

219 allow_recursive: bool = True, 

220 ): 

221 # noinspection GrazieInspection 

222 """ 

223 :param annotation: 用于类型检查的类型 

224 :type annotation: T 

225 :param default: 字段默认值 

226 :type default: Any 

227 :param default_factory: 字段默认值工厂 

228 :type default_factory: Callable[[], Any] | UnsetType 

229 :param allow_recursive: 是否允许递归处理字段值 

230 :type allow_recursive: bool 

231 

232 .. versionchanged:: 0.2.0 

233 重命名参数 ``type_`` 为 ``value`` 

234 

235 重命名参数 ``annotation`` 为 ``default`` 

236 

237 添加参数 ``default_factory`` 

238 """ # noqa: D205 

239 kwargs: dict[str, Any] = {} 

240 if default is not Unset: 

241 kwargs["default"] = default 

242 if default_factory is not Unset: 

243 kwargs["default_factory"] = default_factory 

244 

245 if len(kwargs) != 1: 

246 msg = "take one of arguments 'default' or 'default_factory'" 

247 raise ValueError(msg) 

248 

249 value = default 

250 if not isinstance(default, FieldInfo): 

251 value = FieldInfo(**kwargs) 

252 

253 self.annotation = annotation 

254 self.value = value 

255 self.allow_recursive = allow_recursive 

256 

257 annotation: T 

258 """ 

259 用于类型检查的类型 

260 """ 

261 value: FieldInfo 

262 """ 

263 字段值 

264 """ 

265 allow_recursive: bool 

266 """ 

267 是否允许递归处理字段值 

268 

269 .. versionadded:: 0.1.4 

270 """ 

271 

272 

273class MappingType(BaseModel): 

274 value: type[Mapping] # type: ignore[type-arg] 

275 

276 

277class NestedMapping(BaseModel): 

278 value: Mapping[str, Any] 

279 

280 

281def _is_mapping(typ: Any) -> bool: 

282 """ 

283 判断是否为 :py:class`~collections.abc.Mapping` 类型 

284 

285 :param typ: 待检测类型 

286 :type typ: Any 

287 

288 :return: 是否为 :py:class`~collections.abc.Mapping` 类型 

289 :rtype: bool 

290 """ 

291 if typ is Any: 

292 return True 

293 try: 

294 MappingType(value=typ) 

295 except (ValidationError, TypeError): 

296 return False 

297 return True 

298 

299 

300def _allow_recursive(typ: Any) -> bool: 

301 """ 

302 判断是否允许递归处理字段值(键全为字符串则视为允许) 

303 

304 :param typ: 待检测值 

305 :type typ: Any 

306 

307 :return: 是否允许递归处理字段值 

308 :rtype: bool 

309 """ # noqa: RUF002 

310 try: 

311 NestedMapping(value=typ) 

312 except (ValidationError, TypeError): 

313 return False 

314 return True 

315 

316 

317def _check_overwriting_exists_path( 

318 key: str, value: Any, fmt_data: MappingConfigData[Any], typehint_types: tuple[type, ...] 

319) -> bool: 

320 """ 

321 检查是否覆盖了验证器已存在的路径 

322 

323 :param key: 路径 

324 :type key: str 

325 :param value: 新值 

326 :type value: Any 

327 :param fmt_data: 验证器 

328 :type fmt_data: MappingConfigData[Any] 

329 :param typehint_types: 类型提示类型 

330 :type typehint_types: tuple[type, ...] 

331 

332 .. versionadded:: 0.3.0 

333 """ 

334 # 如果传入了任意路径的父路径 

335 if key not in fmt_data: 

336 return False 

337 

338 # 那就检查新值和旧值是否都为Mapping子类或Any 

339 target_value = fmt_data.retrieve(key) 

340 if not issubclass(type(target_value), typehint_types): 

341 target_value = type(target_value) 

342 

343 # 如果是那就把父路径直接加入parent_set不进行后续操作 

344 if _is_mapping(value) and _is_mapping(target_value): 

345 return True 

346 

347 # 否则发出警告提示意外地复写验证器路径 

348 warnings.warn( 

349 f"Overwriting exists validator path with unexpected type '{value}'(new) and '{target_value}'(exists)", 

350 stacklevel=2, 

351 ) 

352 return False 

353 

354 

355@overload 

356def _convert2definition[D: FieldDefinition[Any]](value: D, typehint_types: tuple[type, ...]) -> D: ... 

357 

358 

359@overload 

360def _convert2definition(value: FieldInfo, typehint_types: tuple[type, ...]) -> FieldDefinition[Any]: ... 

361 

362 

363@overload 

364def _convert2definition[T: AnyTypeHint](value: T, typehint_types: tuple[type, ...]) -> FieldDefinition[T]: ... 

365 

366 

367def _convert2definition(value: Any, typehint_types: tuple[type, ...]) -> FieldDefinition[Any]: 

368 """ 

369 将键值换为字段定义 

370 

371 :param value: 键 

372 :type value: Any 

373 :param typehint_types: 类型提示类型 

374 :type typehint_types: tuple[type, ...] 

375 

376 .. versionadded:: 0.3.0 

377 """ 

378 # foo = FieldInfo() # noqa: ERA001 

379 if isinstance(value, FieldInfo): 

380 # foo: FieldInfo().annotation = FieldInfo() # noqa: ERA001 

381 return FieldDefinition(value.annotation, value) 

382 # foo: int # noqa: ERA001 

383 # 如果是仅类型就填上空值 

384 if issubclass(type(value), typehint_types): 

385 # foo: int = FieldInfo() # noqa: ERA001 

386 return FieldDefinition(value, FieldInfo()) 

387 # foo = FieldDefinition(int, FieldInfo()) # noqa: ERA001 

388 # 已经是处理好的字段定义不需要特殊处理 

389 if isinstance(value, FieldDefinition): 

390 return value 

391 # foo = 1 # noqa: ERA001 

392 # 如果是仅默认值就补上类型 

393 # foo: int = 1 # noqa: ERA001 

394 return FieldDefinition(type(value), FieldInfo(default=value)) 

395 

396 

397class DefaultValidatorFactory[D: MCD]: 

398 """默认的验证器工厂""" 

399 

400 def __init__(self, validator: Iterable[str] | Mapping[str, Any], validator_options: ValidatorOptions): 

401 # noinspection GrazieInspection 

402 """ 

403 :param validator: 用于生成验证器的数据 

404 :type validator: Iterable[str] | Mapping[str, Any] 

405 :param validator_options: 验证器选项 

406 :type validator_options: ValidatorOptions 

407 

408 额外验证器选项 

409 ----------------------- 

410 .. list-table:: 

411 :widths: auto 

412 

413 * - 键名 

414 - 描述 

415 - 默认值 

416 - 类型 

417 * - model_config_key 

418 - 内部编译 :py:mod:`pydantic` 的 :py:class:`~pydantic.main.BaseModel` 

419 时,模型配置是以嵌套字典的形式存储的,因此请确保此参数不与任何其中子模型名冲突 

420 - ".__model_config__" 

421 - Any 

422 

423 .. versionchanged:: 0.1.2 

424 支持验证器混搭路径字符串和嵌套字典 

425 

426 .. versionchanged:: 0.1.4 

427 支持验证器非字符串键 (含有非字符串键的子验证器不会被递归处理) 

428 """ # noqa: RUF002, D205 

429 validator = deepcopy(validator) 

430 if isinstance(validator, Mapping): # 先检查Mapping因为Mapping可以是Iterable 

431 ... 

432 elif isinstance(validator, Iterable): 

433 # 预处理为 

434 # k: Any # noqa: ERA001 

435 validator = OrderedDict((k, Any) for k in validator) 

436 else: 

437 msg = f"Invalid validator type '{type(validator).__name__}'" 

438 raise TypeError(msg) 

439 self.validator = validator 

440 self.validator_options = validator_options 

441 

442 self.typehint_types = (type, types.UnionType, types.EllipsisType, types.GenericAlias, TypeAliasType) 

443 self.model_config_key = validator_options.extra.get("model_config_key", ".__model_config__") 

444 self._compile() 

445 self.model: type[BaseModel] 

446 

447 def _fmt_mapping_key(self, validator: Mapping[str, Any]) -> tuple[Mapping[str, Any], set[str | ABCPath[Any]]]: 

448 # noinspection GrazieInspection 

449 """ 

450 格式化验证器键 

451 

452 :param validator: Mapping验证器 

453 :type validator: Mapping[str, Any] 

454 

455 :return: 格式化后的映射键和被覆盖的Mapping父路径 

456 :rtype: tuple[Mapping[str, Any], set[str | ABCPath[Any]]] 

457 

458 .. versionchanged:: 0.3.0 

459 拆分覆盖检查到函数 :py:func:`_check_overwriting_exists_path` 

460 """ 

461 iterator = iter(validator.items()) 

462 key: str = None # type: ignore[assignment] 

463 value: Any = None 

464 

465 def _next() -> bool: 

466 """ 

467 获取下一个键值对 

468 

469 :return: 是否耗尽迭代器 

470 :rtype: bool 

471 """ 

472 nonlocal key, value 

473 with suppress(StopIteration): 

474 key, value = next(iterator) 

475 return False 

476 return True 

477 

478 # 如果为空则提前返回 

479 if _next(): 

480 return {}, set() 

481 

482 fmt_data: MappingConfigData[OrderedDict[str, Any]] = MappingConfigData(OrderedDict()) 

483 parent_set: set[str | ABCPath[Any]] = set() 

484 while True: 

485 # 如果传入了任意路径的父路径那就检查新值和旧值是否都为Mapping子类或Any 

486 # 如果是那就把父路径直接加入parent_set不进行后续操作 

487 if _check_overwriting_exists_path(key, value, fmt_data, self.typehint_types): 

488 parent_set.add(key) 

489 if _next(): # 更新键值对 

490 break 

491 continue 

492 

493 # 如果可以递归处理字段值那就递归处理 

494 if _allow_recursive(value): 

495 value, inner_path_set = self._fmt_mapping_key(value) 

496 parent_set.update(f"{key}\\.{inner_path}" for inner_path in inner_path_set) 

497 

498 # 记录该键值对 

499 try: 

500 fmt_data.modify(key, value) 

501 except ConfigDataTypeError as err: # 如果任意父路径不为Mapping 

502 relative_path = Path(err.key_info.relative_keys) 

503 # 如果旧类型为Mapping子类或Any那么就允许新的键创建 

504 if not _is_mapping(fmt_data.retrieve(relative_path)): 

505 raise err from None 

506 fmt_data.modify(relative_path, OrderedDict()) 

507 parent_set.add(relative_path) 

508 fmt_data.modify(key, value) # 再次记录该键值对 

509 

510 # 获取下一个键值对 

511 if _next(): 

512 break 

513 

514 return fmt_data.data, parent_set 

515 

516 def _mapping2model(self, mapping: Mapping[str, Any], model_config: dict[str, Any]) -> type[BaseModel]: 

517 """ 

518 将Mapping转换为Model 

519 

520 :param mapping: 需要转换的Mapping 

521 :type mapping: Mapping[str, Any] 

522 

523 :return: 转换后的Model 

524 :rtype: type[BaseModel] 

525 

526 .. versionchanged:: 0.3.0 

527 拆分字段定义转换到函数 :py:func:`_convert2definition` 

528 """ 

529 fmt_data: OrderedDict[str, Any] = OrderedDict() 

530 for key, value in mapping.items(): 

531 # 将键值对转换为字段定义 

532 definition = _convert2definition(value, self.typehint_types) 

533 

534 # 递归处理Mapping值 

535 if all( 

536 ( 

537 definition.allow_recursive, 

538 _allow_recursive(definition.value.default), 

539 # foo.bar = {} # noqa: ERA001 

540 # 这种情况下不进行递归解析 即捕获所有键(foo.bar.*)如果进行了解析就会忽略所有内容即返回foo.bar={} 

541 definition.value.default, 

542 ) 

543 ): 

544 model_cls = self._mapping2model( 

545 mapping=definition.value.default, model_config=model_config.get(key, {}) 

546 ) 

547 definition = FieldDefinition(model_cls, FieldInfo(default_factory=model_cls)) 

548 

549 # 如果忽略不存在的键则填充特殊值 

550 if all((self.validator_options.skip_missing, definition.value.is_required())): 

551 definition = FieldDefinition(definition.annotation | SkipMissingType, FieldInfo(default=SkipMissing)) 

552 

553 fmt_data[key] = (definition.annotation, definition.value) 

554 

555 # 创建验证模型 

556 # noinspection PyInvalidCast 

557 return create_model( 

558 f"{type(self).__name__}.RuntimeTemplate", 

559 __config__=cast(ConfigDict, model_config.get(self.model_config_key, {})), 

560 **fmt_data, 

561 ) 

562 

563 def _compile(self) -> None: 

564 """编译模板""" 

565 fmt_validator, parent_set = self._fmt_mapping_key(self.validator) 

566 # 所有重复存在的父路径都将允许其下存在多余的键 

567 model_config: MCD = MappingConfigData() 

568 for path in parent_set: 

569 model_config.modify(path, {self.model_config_key: {"extra": "allow"}}) 

570 

571 self.model = self._mapping2model(fmt_validator, model_config.data) 

572 

573 # noinspection PyTypeHints 

574 def __call__(self, config_ref: Ref[D | NoneConfigData]) -> D: 

575 """ 

576 验证配置数据 

577 

578 :param config_ref: 配置数据引用 

579 :type config_ref: Ref[D | NoneConfigData] 

580 

581 :return: 验证后的配置数据 

582 :rtype: D 

583 """ 

584 if isinstance(config_ref.value, NoneConfigData): 

585 config_ref.value = MappingConfigData() # type: ignore[assignment] 

586 data: D = config_ref.value # type: ignore[assignment] 

587 

588 try: 

589 dict_obj = self.model(**data.data).model_dump() 

590 except ValidationError as err: 

591 raise _process_pydantic_exceptions(err) from err 

592 

593 # 处理 SkipMissing 项 

594 if self.validator_options.skip_missing: 

595 dict_obj = _remove_skip_missing(dict_obj) 

596 

597 # 完全替换原始数据 

598 if self.validator_options.allow_modify: 

599 data._data = dict_obj # noqa: SLF001 

600 return data 

601 return data.from_data(dict_obj) 

602 

603 

604# noinspection PyTypeHints 

605def pydantic_validator[D: MCD]( 

606 validator: type[BaseModel], cfg: ValidatorOptions 

607) -> Callable[[Ref[D | NoneConfigData]], D]: 

608 """ 

609 验证器选项 ``skip_missing`` 无效 

610 

611 :param validator: :py:class:`~pydantic.main.BaseModel` 的子类 

612 :type validator: type[BaseModel] 

613 :param cfg: 验证器选项 

614 :type cfg: ValidatorOptions 

615 

616 :return: 验证器 

617 :rtype: Callable[[Ref[D | NoneConfigData]], D] 

618 """ 

619 if not issubclass(validator, BaseModel): 

620 msg = f"Expected a subclass of BaseModel for parameter 'validator', but got '{validator.__name__}'" 

621 raise TypeError(msg) 

622 if cfg.skip_missing: 

623 warnings.warn("skip_missing is not supported in pydantic validator", stacklevel=2) 

624 

625 # noinspection PyTypeHints 

626 def _builder(config_ref: Ref[D | NoneConfigData]) -> D: 

627 """ 

628 验证配置数据 

629 

630 :param config_ref: 配置数据引用 

631 :type config_ref: Ref[D | NoneConfigData] 

632 

633 :return: 验证后的配置数据 

634 :rtype: D 

635 """ 

636 if isinstance(config_ref.value, NoneConfigData): 

637 config_ref.value = MappingConfigData() # type: ignore[assignment] 

638 data: D = config_ref.value # type: ignore[assignment] 

639 

640 try: 

641 dict_obj = validator(**data).model_dump() 

642 except ValidationError as err: 

643 raise _process_pydantic_exceptions(err) from err 

644 

645 # 完全替换原始数据 

646 if cfg.allow_modify: 

647 data._data = dict_obj # noqa: SLF001 

648 return data 

649 return data.from_data(dict_obj) 

650 

651 return _builder 

652 

653 

654class ComponentValidatorFactory[D: ComponentConfigData[Any, Any]]: 

655 """ 

656 组件验证器工厂 

657 

658 .. versionadded:: 0.2.0 

659 """ 

660 

661 def __init__(self, validator: Mapping[str | None, Callable[[Ref[ICD]], ICD]], validator_options: ValidatorOptions): 

662 """ 

663 :param validator: 组件验证器 

664 :type validator: Mapping[str | None, Callable[[Ref[ICD]], ICD]] 

665 :param validator_options: 验证器选项 

666 :type validator_options: ValidatorOptions 

667 

668 额外验证器选项 

669 ----------------------- 

670 

671 .. list-table:: 

672 :widths: auto 

673 

674 * - 键名 

675 - 描述 

676 - 默认值 

677 - 类型 

678 * - allow_initialize 

679 - 是否允许初始化不存在的组件成员(注意! 现在的实现方式会强制初始化成员为 :py:class:`MappingConfigData`) 

680 - True 

681 - bool 

682 * - meta_validator 

683 - 组件元数据验证器 

684 - 尝试从传入的组件元数据获得,若不存在(值为None)则放弃验证 

685 - Callable[[ComponentMeta, ValidatorOptions], ComponentMeta] 

686 

687 .. versionchanged:: 0.3.0 

688 更改参数 ``validator`` 类型为 ``Mapping[str | None, Callable[[Ref[ICD]], ICD]]`` 

689 并移除因此冗余的移除额外验证器选项 ``validator_factory`` 

690 """ # noqa: RUF002, D205 

691 self.validator_options = validator_options 

692 self.validators = validator 

693 

694 def _validate_member_metadata(self, component_data: D) -> dict[str, ICD]: 

695 """ 

696 验证组件成员元数据 

697 

698 :param component_data: 组件数据 

699 :type component_data: D 

700 

701 :return: 验证后的组件数据 

702 :rtype: dict[str | None, ICD] 

703 """ 

704 validated_members: dict[str, ICD] = {} 

705 for member, validator in self.validators.items(): 

706 if member is None: 

707 continue 

708 

709 member_not_exists = member not in component_data 

710 member_data_ref: Ref[ICD] 

711 if member_not_exists and self.validator_options.extra.get("allow_initialize", True): 

712 member_data_ref = Ref(MappingConfigData()) 

713 if self.validator_options.allow_modify: 

714 component_data[member] = member_data_ref.value 

715 elif member_not_exists: 

716 raise ComponentMemberMismatchError(missing={member}, redundant=set()) 

717 else: 

718 member_data_ref = Ref(component_data[member]) 

719 validated_member = validator(member_data_ref) 

720 validated_members[member] = validated_member 

721 

722 # 完全替换成员数据 

723 if self.validator_options.allow_modify: 

724 component_data[member] = validated_member 

725 return validated_members 

726 

727 def __call__(self, config_ref: Ref[D | NoneConfigData]) -> D: 

728 """ 

729 验证配置数据 

730 

731 :param config_ref: 配置数据引用 

732 :type config_ref: Ref[D | NoneConfigData] 

733 

734 :return: 验证后的配置数据 

735 :rtype: D 

736 

737 .. versionchanged:: 0.3.0 

738 拆分验证成员元数据到方法 :py:meth:`_validate_member_metadata` 

739 """ 

740 if isinstance(config_ref.value, NoneConfigData): 

741 config_ref.value = ComponentConfigData() # type: ignore[assignment] 

742 

743 component_ref: Ref[D] = config_ref # type: ignore[assignment] 

744 component_data = component_ref.value 

745 

746 validated_members: dict[str, ICD] = self._validate_member_metadata(component_data) 

747 

748 meta = deepcopy(component_data.meta) 

749 if None in self.validators: 

750 meta.config = self.validators[None](Ref(meta.config)) 

751 

752 # noinspection PyUnresolvedReferences 

753 meta_validator = None if meta.parser is None else meta.parser.validator 

754 meta_validator = self.validator_options.extra.get("meta_validator", meta_validator) 

755 if meta_validator is not None: 

756 meta = meta_validator(meta, self.validator_options) 

757 

758 # 完全替换元数据 

759 if self.validator_options.allow_modify: 

760 component_data._meta = meta # noqa: SLF001 

761 

762 return component_data.from_data(meta, validated_members) 

763 

764 

765__all__ = ( 

766 "ComponentValidatorFactory", 

767 "DefaultValidatorFactory", 

768 "FieldDefinition", 

769 "ValidatorOptions", 

770 "ValidatorTypes", 

771 "pydantic_validator", 

772)