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

1# cython: language_level = 3 # noqa: ERA001

4"""

5组件配置数据实现

7.. versionadded:: 0.2.0

8"""

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

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

40@dataclass

41class ComponentOrders:

42 """

43 组件顺序

45 .. versionadded:: 0.2.0

46 """

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)

54@dataclass

55class ComponentMember:

56 """

57 组件成员

59 .. versionadded:: 0.2.0

60 """

62 filename: str

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

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

67@dataclass

68class ComponentMeta[D: ABCConfigData]:

69 """

70 组件元数据

72 .. versionadded:: 0.2.0

73 """

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)

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

82 BasicConfigData[D], ABCIndexedConfigData[D]

83):

84 """

85 组件配置数据

87 .. versionadded:: 0.2.0

88 """

90 def __init__(self, meta: M | None = None, members: Mapping[str, D] | None = None): # type: ignore[var-annotated] # mypy抽风

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()

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 """

235 获取路径的值的*快照*

236

237 操作顺序由 :py:attr:`~ComponentOrders.read` 指定

238

239 :param path: 路径

240 :type path: PathLike

241 :param return_raw_value: 是否获取原始值，为 :py:const:`False` 时，会将Mapping | Sequence转换为对应类

242 :type return_raw_value: bool

243

244 :return: 路径的值

245 :rtype: Any

246

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

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

249

250 .. versionchanged:: 0.2.0

251 重命名参数 ``get_raw`` 为 ``return_raw_value``

252 """ # noqa: RUF002

253 path = fmt_path(path)

254

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

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

257

258 return self._resolve_members(

259 path,

260 order=self._meta.orders.read,

261 processor=processor,

262 exception=RequiredPathNotFoundError(

263 key_info=KeyInfo(path, 0),

264 operate=ConfigOperate.Read,

265 ),

266 )

267

268 @override

269 @check_read_only

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

271 # noinspection PyIncorrectDocstring

272 """

273 修改路径的值

274

275 操作顺序由 :py:attr:`~ComponentOrders.update` 先尝试对现有数据进行更新再由

276 :py:attr:`~ComponentOrders.create` 尝试创建新数据

277

278 :param path: 路径

279 :type path: PathLike

280 :param value: 值

281 :type value: Any

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

283 :type allow_create: bool

284

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

286 :rtype: Self

287

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

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

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

291

292 .. caution::

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

294

295 .. attention::

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

297

298 .. versionchanged:: 0.3.0

299 现在正确的由 :py:attr:`~ComponentOrders.update` 先尝试对现有数据进行更新再由

300 :py:attr:`~ComponentOrders.create` 尝试创建新数据

301 """ # noqa: RUF002

302 path = fmt_path(path)

303

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

305 try:

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

307 except (RequiredPathNotFoundError, ConfigDataTypeError) as err:

308 raise RequiredPathNotFoundError(

309 key_info=err.key_info,

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

311 ) from None

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

313

314 with suppress(RequiredPathNotFoundError):

315 self._resolve_members(

316 path,

317 order=self._meta.orders.update,

318 processor=_update_processor,

319 exception=RequiredPathNotFoundError(

320 key_info=KeyInfo(path, 0),

321 operate=ConfigOperate.Write,

322 ),

323 )

324 return self

325

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

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

328

329 self._resolve_members(

330 path,

331 order=self._meta.orders.create,

332 processor=_create_processor,

333 exception=RequiredPathNotFoundError(

334 key_info=KeyInfo(path, 0),

335 operate=ConfigOperate.Write,

336 ),

337 )

338 return self

339

340 @override

341 @check_read_only

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

343 """

344 删除路径

345

346 操作顺序由 :py:attr:`~ComponentOrders.delete` 指定

347

348 :param path: 路径

349 :type path: PathLike

350

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

352 :rtype: Self

353

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

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

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

357 """

358 path = fmt_path(path)

359

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

361 # noinspection PyArgumentList

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

363

364 self._resolve_members(

365 path,

366 order=self._meta.orders.delete,

367 processor=processor,

368 exception=RequiredPathNotFoundError(

369 key_info=KeyInfo(path, 0),

370 operate=ConfigOperate.Delete,

371 ),

372 )

373 return self

374

375 @override

376 @check_read_only

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

378 """

379 确保路径不存在 (删除路径，但是找不到路径时不会报错)

380

381 操作顺序由 :py:attr:`~ComponentOrders.delete` 指定

382

383 :param path: 路径

384 :type path: PathLike

385

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

387 :rtype: Self

388

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

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

391

392 .. versionadded:: 0.1.2

393 """ # noqa: RUF002

394 path = fmt_path(path)

395

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

397 # noinspection PyArgumentList

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

399

400 with suppress(RequiredPathNotFoundError):

401 self._resolve_members(

402 path,

403 order=self._meta.orders.delete,

404 processor=processor,

405 exception=RequiredPathNotFoundError(

406 key_info=KeyInfo(path, 0),

407 operate=ConfigOperate.Delete,

408 ),

409 )

410 return self

411

412 @override

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

414 """

415 判断路径是否存在

416

417 操作顺序由 :py:attr:`~ComponentOrders.read` 指定

418

419 :param path: 路径

420 :type path: PathLike

421 :param ignore_wrong_type: 忽略配置数据类型错误

422 :type ignore_wrong_type: bool

423

424 :return: 路径是否存在

425 :rtype: bool

426

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

428 """

429 if not self._meta.orders.read:

430 return False

431 path = fmt_path(path)

432

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

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

435

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

437 return self._resolve_members(

438 path,

439 order=self._meta.orders.read,

440 processor=processor,

441 exception=RequiredPathNotFoundError(

442 key_info=KeyInfo(path, 0),

443 operate=ConfigOperate.Delete,

444 ),

445 )

446 return False

447

448 @override

449 def get[V](

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

451 ) -> V | Any:

452 """

453 获取路径的值的*快照*，路径不存在时填充默认值

454

455 操作顺序由 :py:attr:`~ComponentOrders.read` 指定

456

457 :param path: 路径

458 :type path: PathLike

459

460 :param default: 默认值

461 :type default: V

462 :param return_raw_value: 是否获取原始值

463 :type return_raw_value: bool

464

465 :return: 路径的值

466 :rtype: V | Any

467

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

469

470 .. versionchanged:: 0.2.0

471 重命名参数 ``get_raw`` 为 ``return_raw_value``

472 """ # noqa: RUF002

473 path = fmt_path(path)

474

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

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

477

478 with suppress(RequiredPathNotFoundError):

479 return self._resolve_members(

480 path,

481 order=self._meta.orders.read,

482 processor=processor,

483 exception=RequiredPathNotFoundError(

484 key_info=KeyInfo(path, 0),

485 operate=ConfigOperate.Read,

486 ),

487 )

488 return default

489

490 @override

491 @check_read_only

492 def setdefault[V](

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

494 ) -> V | Any:

495 """

496 如果路径不在配置数据中则填充默认值到配置数据并返回

497

498 操作顺序由 :py:attr:`~ComponentOrders.read` 先尝试对现有数据进行读取再由

499 :py:attr:`~ComponentOrders.create` 尝试创建新数据

500

501 :param path: 路径

502 :type path: PathLike

503 :param default: 默认值

504 :type default: V

505 :param return_raw_value: 是否获取原始值

506 :type return_raw_value: bool

507

508 :return: 路径的值

509 :rtype: V | Any

510

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

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

513

514 .. versionchanged:: 0.2.0

515 重命名参数 ``get_raw`` 为 ``return_raw_value``

516

517 重命名 ``set_default`` 为 ``setdefault``

518 """

519 path = fmt_path(path)

520

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

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

523

524 with suppress(RequiredPathNotFoundError):

525 return self._resolve_members(

526 path,

527 order=self._meta.orders.read,

528 processor=_retrieve_processor,

529 exception=RequiredPathNotFoundError(

530 key_info=KeyInfo(path, 0),

531 operate=ConfigOperate.Read,

532 ),

533 )

534

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

536 member.modify(pth, default)

537 return default

538

539 return self._resolve_members(

540 path,

541 order=self._meta.orders.create,

542 processor=_modify_processor,

543 exception=RequiredPathNotFoundError(

544 key_info=KeyInfo(path, 0),

545 operate=ConfigOperate.Write,

546 ),

547 )

548

549 @override

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

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

552 return NotImplemented

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

554

555 __hash__ = None # type: ignore[assignment]

556

557 @override

558 def __str__(self) -> str:

559 return str(self._members)

560

561 @override

562 def __repr__(self) -> str:

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

564

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

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

567

568 @override

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

570 return key in self._members

571

572 @override

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

574 return iter(self._members)

575

576 @override

577 def __len__(self) -> int:

578 return len(self._members)

579

580 @override

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

582 return self._members[index]

583

584 @override

585 @check_read_only

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

587 """

588 .. danger::

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

590 """ # noqa: RUF002, D205

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

592

593 @override

594 @check_read_only

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

596 """

597 .. danger::

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

599 """ # noqa: RUF002, D205

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

601

602

603__all__ = (

604 "ComponentConfigData",

605 "ComponentMember",

606 "ComponentMeta",

607 "ComponentOrders",

608)