Coverage for configzen/routes.py: 79%

1"""Routes creation and parsing.""" 

2 

3from __future__ import annotations 8 ctx1abcdeghf

4 

5from functools import reduce, singledispatchmethod 8 ctx1abcdeghf

6from typing import ( 8 ctx1abcdeghf

7 TYPE_CHECKING, 

8 Any, 

9 ClassVar, 

10 Generic, 

11 Type, 

12 TypeVar, 

13 Union, 

14 get_origin, 

15) 

16 

17from class_singledispatch import class_singledispatch 8 ctx1abcdeghf

18 

19from configzen.errors import LinkedRouteError, RouteError 8 ctx1abcdeghf

20from configzen.typedefs import ConfigObject 8 ctx1abcdeghf

21 

22if TYPE_CHECKING: 8 ctx1abcdeghf

23 from collections.abc import Iterator 

24 

25 from typing_extensions import Self, TypeAlias 

26 

27 

28__all__ = ( 8 ctx1abcdeghf

29 "GetAttr", 

30 "GetItem", 

31 "Route", 

32 "RouteLike", 

33 "Step", 

34) 

35 

36RouteLike: TypeAlias = Union[ 8 ctx1abcdeghf

37 str, 

38 int, 

39 "list[str | int]", 

40 "tuple[str | int, ...]", 

41 "list[Step[Any]]", 

42 "tuple[Step[Any], ...]", 

43 "Route", 

44 "Step[Any]", 

45] 

46 

47_KT = TypeVar("_KT") 8 ctx1abcdeghf

48 

49 

50class Step(Generic[_KT]): 8 ctx1abcdeghf

51 """ 

52 A configuration route step. 

53 

54 Do not use this class directly. Use GetAttr or GetItem instead. 

55 """ 

56 

57 key: _KT 8 ctx1abcdeghf

58 

59 def __init__(self, key: _KT, /) -> None: 8 ctx1abcdeghf

60 self.key = key 8 ctx1abcdeghf

61 

62 def __eq__(self, other: object) -> bool: 8 ctx1abcdeghf

63 """Compare this step to another step.""" 

64 if isinstance(other, Step): 8 ctx1abcdeghf

65 return ( 8 ctx1abcdeghf

66 issubclass(type(other), type(self)) 

67 or issubclass(type(self), type(other)) 

68 ) or self.key == other.key 

69 return NotImplemented 

70 

71 def get(self, _: Any, /) -> object: 8 ctx1abcdeghf

72 """Perform a get operation.""" 

73 raise NotImplementedError 

74 

75 def set(self, _: Any, __: object, /) -> None: 8 ctx1abcdeghf

76 """Perform a set operation.""" 

77 raise NotImplementedError 

78 

79 def __call__(self, obj: Any, /) -> object: 8 ctx1abcdeghf

80 """Perform a get operation.""" 

81 return self.get(obj) 

82 

83 def __repr__(self) -> str: 8 ctx1abcdeghf

84 """Represent this step in a string.""" 

85 return f"{type(self).__name__}({self.key!r})" 8 ctx1abcdeghf

86 

87 

88class GetAttr(Step[str]): 8 ctx1abcdeghf

89 """ 

90 A route step that gets an attribute from an object. 

91 

92 The argument is used as an attribute name. 

93 """ 

94 

95 def get(self, target: Any, /) -> object: 8 ctx1abcdeghf

96 """Get an attribute from an object.""" 

97 return getattr(target, self.key) 

98 

99 def set(self, target: Any, value: object, /) -> None: 8 ctx1abcdeghf

100 """Set an attribute in an object.""" 

101 setattr(target, self.key, value) 

102 

103 def __str__(self) -> str: 8 ctx1abcdeghf

104 """Compose this step into a string.""" 

105 return str(self.key).replace(Route.TOKEN_DOT, r"\.") 8 ctx1abcdeghf

106 

107 

108class GetItem(Step[Union[int, str]]): 8 ctx1abcdeghf

109 r""" 

110 A route step that gets an item from an object. 

111 

112 If the argument is a string, it is used checked for being a digit. 

113 Unless explicitly escaped, if it is a digit, it is casted to an integer. 

114 Otherwise, it is used as is. 

115 """ 

116 

117 def __init__(self, key: int | str, /, *, ignore_digit: bool = False) -> None: 8 ctx1abcdeghf

118 self.escape = False 8 ctx1abcdeghf

119 if isinstance(key, str) and key.isdigit(): 8 ctx1abcdeghf

120 if ignore_digit: 120 ↛ 121line 120 didn't jump to line 121, because the condition on line 120 was never true8 ctx1abcdeghf

121 self.escape = True 

122 else: 

123 key = int(key) 8 ctx1abcdeghf

124 super().__init__(key) 8 ctx1abcdeghf

125 

126 def get(self, target: Any, /) -> object: 8 ctx1abcdeghf

127 """Get an item from an object.""" 

128 return target[self.key] 

129 

130 def set(self, target: Any, value: object, /) -> None: 8 ctx1abcdeghf

131 """Set an item in an object.""" 

132 target[self.key] = value 

133 

134 def __str__(self) -> str: 8 ctx1abcdeghf

135 """Compose this step into a string.""" 

136 argument = str(self.key) 8 ctx1abcdeghf

137 if self.escape: 137 ↛ 138line 137 didn't jump to line 138, because the condition on line 137 was never true8 ctx1abcdeghf

138 argument = Route.TOKEN_ESCAPE + argument 

139 return argument.join( 8 ctx1abcdeghf

140 Route.TOKEN_ENTER + Route.TOKEN_LEAVE, 

141 ).replace(Route.TOKEN_DOT, r"\.") 

142 

143 

144def _route_decompose( # noqa: C901, PLR0912, PLR0915 8 ctx1abcdeghf

145 route: str, 

146 *, 

147 dot: str, 

148 escape: str, 

149 enter: str, 

150 leave: str, 

151) -> list[Step[Any]]: 

152 """ 

153 Decompose a route into a list of steps. 

154 

155 Parameters 

156 ---------- 

157 route 

158 A route to decompose. 

159 dot 

160 A token used to separate steps. 

161 escape 

162 A token used to escape a token. 

163 enter 

164 A token used to enter an item. 

165 leave 

166 A token used to exit an item. 

167 

168 Returns 

169 ------- 

170 List of steps. 

171 

172 """ 

173 if not route.endswith(dot): 173 ↛ 176line 173 didn't jump to line 176, because the condition on line 173 was always true8 ctx1abcdeghf

174 route += dot 8 ctx1abcdeghf

175 

176 key = "" 8 ctx1abcdeghf

177 entered: int | None = None 8 ctx1abcdeghf

178 left: bool = False 8 ctx1abcdeghf

179 steps: list[Step[Any]] = [] 8 ctx1abcdeghf

180 emit = steps.append 8 ctx1abcdeghf

181 escaping = False 8 ctx1abcdeghf

182 escaped = False 8 ctx1abcdeghf

183 step: Step[Any] 

184 

185 for index, token in enumerate(route): 8 ctx1abcdeghf

186 if escaping: 8 ctx1abcdeghf

187 key += token 8 ctx1abcdeghf

188 escaping = False 8 ctx1abcdeghf

189 escaped = True 8 ctx1abcdeghf

190 continue 8 ctx1abcdeghf

191 is_last = index == len(route) - 1 8 ctx1abcdeghf

192 if token == dot: 8 ctx1abcdeghf

193 if entered is not None: 193 ↛ 194line 193 didn't jump to line 194, because the condition on line 193 was never true8 ctx1abcdeghf

194 key += token 

195 else: 

196 if left: 8 ctx1abcdeghf

197 step = GetItem(key, ignore_digit=escaped) 8 ctx1abcdeghf

198 left = False 8 ctx1abcdeghf

199 else: 

200 step = GetAttr(key) 8 ctx1abcdeghf

201 escaped = False 8 ctx1abcdeghf

202 emit(step) 8 ctx1abcdeghf

203 key = "" 8 ctx1abcdeghf

204 elif token == escape: 8 ctx1abcdeghf

205 if is_last: 205 ↛ 206line 205 didn't jump to line 206, because the condition on line 205 was never true8 ctx1abcdeghf

206 key += token 

207 else: 

208 escaping = True 8 ctx1abcdeghf

209 elif token == enter: 8 ctx1abcdeghf

210 if entered is not None: 210 ↛ 211line 210 didn't jump to line 211, because the condition on line 210 was never true8 ctx1abcdeghf

211 msg = f"Already seen {enter!r} that was not closed with {leave!r}" 

212 raise RouteError(msg, route=route, index=index) 

213 if key or index == 0: 213 ↛ 225line 213 didn't jump to line 225, because the condition on line 213 was always true8 ctx1abcdeghf

214 entered = index 8 ctx1abcdeghf

215 if index: 215 ↛ 235line 215 didn't jump to line 235, because the condition on line 215 was always true8 ctx1abcdeghf

216 if left: 8 ctx1abcdeghf

217 step = GetItem(key, ignore_digit=escaped) 8 ctx1abcdeghf

218 left = False 8 ctx1abcdeghf

219 else: 

220 step = GetAttr(key) 8 ctx1abcdeghf

221 escaped = False 8 ctx1abcdeghf

222 emit(step) 8 ctx1abcdeghf

223 key = "" 8 ctx1abcdeghf

224 else: 

225 msg = f"No key between {route[index-1]!r} and {token!r}" 

226 raise RouteError(msg, route=route, index=index) 

227 elif token == leave: 8 ctx1abcdeghf

228 if entered is None: 228 ↛ 229line 228 didn't jump to line 229, because the condition on line 228 was never true8 ctx1abcdeghf

229 msg = f"{token!r} not preceded by {enter!r} token" 

230 raise RouteError(msg, route=route, index=index) 

231 entered = None 8 ctx1abcdeghf

232 left = True 8 ctx1abcdeghf

233 else: 

234 key += token 8 ctx1abcdeghf

235 if is_last and entered is not None: 235 ↛ 236line 235 didn't jump to line 236, because the condition on line 235 was never true8 ctx1abcdeghf

236 msg = f"Expected {leave!r} token" 

237 raise RouteError(msg, route=route, index=entered) 

238 return steps 8 ctx1abcdeghf

239 

240 

241class Route: 8 ctx1abcdeghf

242 r""" 

243 Routes are, lists of steps that are used to access values in a configuration. 

244 

245 Each step is either a key or an index. 

246 

247 A route can be created from a string, a list of steps, or another route. 

248 

249 Examples 

250 -------- 

251 >>> route = Route("a.b.c") 

252 >>> route 

253 <Route 'a.b.c'> 

254 >>> route.steps 

255 [GetAttr('a'), GetAttr('b'), GetAttr('c')] 

256 

257 Parameters 

258 ---------- 

259 route 

260 A route to parse. 

261 allow_empty 

262 Whether to allow empty routes. 

263 

264 """ 

265 

266 TOKEN_DOT: ClassVar[str] = "." 8 ctx1abcdeghf

267 TOKEN_ESCAPE: ClassVar[str] = "\\" 8 ctx1abcdeghf

268 TOKEN_ENTER: ClassVar[str] = "[" 8 ctx1abcdeghf

269 TOKEN_LEAVE: ClassVar[str] = "]" 8 ctx1abcdeghf

270 

271 TOKENS: ClassVar[tuple[str, str, str, str]] = ( 8 ctx1abcdeghf

272 TOKEN_DOT, 

273 TOKEN_ESCAPE, 

274 TOKEN_ENTER, 

275 TOKEN_LEAVE, 

276 ) 

277 

278 def __init__( 8 ctx1abcdeghf

279 self, 

280 route: RouteLike, 

281 *, 

282 allow_empty: bool = False, 

283 ) -> None: 

284 steps = self.parse(route) 8 ctx1abcdeghf

285 if not (allow_empty or steps): 285 ↛ 286line 285 didn't jump to line 286, because the condition on line 285 was never true8 ctx1abcdeghf

286 msg = "Empty configuration route" 

287 raise ValueError(msg) 

288 self.__steps = tuple(steps) 8 ctx1abcdeghf

289 

290 @property 8 ctx1abcdeghf

291 def steps(self) -> list[Step[Any]]: 8 ctx1abcdeghf

292 """Get all steps in this route.""" 

293 return list(self.__steps) 8 ctx1abcdeghf

294 

295 def __hash__(self) -> int: 8 ctx1abcdeghf

296 """Get a hash of this route.""" 

297 return hash(self.__steps) 

298 

299 @classmethod 8 ctx1abcdeghf

300 def parse(cls, route: RouteLike) -> list[Step[Any]]: 8 ctx1abcdeghf

301 """ 

302 Parse a route into steps. 

303 

304 Parameters 

305 ---------- 

306 route 

307 The route to parse. 

308 

309 Returns 

310 ------- 

311 List of steps. 

312 

313 """ 

314 if isinstance(route, Step): 8 ctx1abcdeghf

315 return [route] 8 ctx1abcdeghf

316 if isinstance(route, Route): 316 ↛ 317line 316 didn't jump to line 317, because the condition on line 316 was never true8 ctx1abcdeghf

317 return route.steps 

318 if isinstance(route, (tuple, list)): 8 ctx1abcdeghf

319 patched_route: list[Step[Any]] = [] 8 ctx1abcdeghf

320 for element in route: 8 ctx1abcdeghf

321 if isinstance(element, (str, int)): 321 ↛ 322line 321 didn't jump to line 322, because the condition on line 321 was never true8 ctx1abcdeghf

322 try: 

323 patched_element = next(iter(cls.parse(element))) 

324 except StopIteration: 

325 continue 

326 else: 

327 patched_element = element 8 ctx1abcdeghf

328 patched_route.append(patched_element) 8 ctx1abcdeghf

329 return patched_route 8 ctx1abcdeghf

330 if isinstance(route, int): 330 ↛ 331line 330 didn't jump to line 331, because the condition on line 330 was never true8 ctx1abcdeghf

331 return [GetItem(route)] 

332 if isinstance(route, str): 332 ↛ 334line 332 didn't jump to line 334, because the condition on line 332 was always true8 ctx1abcdeghf

333 return cls.decompose(route) 8 ctx1abcdeghf

334 msg = f"Invalid route type {type(route)!r}" 

335 raise TypeError(msg) 

336 

337 @classmethod 8 ctx1abcdeghf

338 def decompose(cls, route: str) -> list[Step[Any]]: 8 ctx1abcdeghf

339 """ 

340 Decompose a route into a list of steps. 

341 

342 Parameters 

343 ---------- 

344 route 

345 A route to decompose. 

346 

347 Returns 

348 ------- 

349 List of steps. 

350 

351 """ 

352 if not route: 8 ctx1abcdeghf

353 return [] 8 ctx1abcdeghf

354 

355 dot, escape, enter, leave = cls.TOKENS 8 ctx1abcdeghf

356 

357 return _route_decompose( 8 ctx1abcdeghf

358 route, 

359 dot=dot, 

360 escape=escape, 

361 enter=enter, 

362 leave=leave, 

363 ) 

364 

365 def compose(self) -> str: 8 ctx1abcdeghf

366 """Compose this route into a string.""" 

367 composed = "" 8 ctx1abcdeghf

368 steps = self.__steps 8 ctx1abcdeghf

369 for index, step in enumerate(steps): 8 ctx1abcdeghf

370 composed += str(step) 8 ctx1abcdeghf

371 if index < len(steps) - 1: 8 ctx1abcdeghf

372 ahead = steps[index + 1] 8 ctx1abcdeghf

373 if isinstance(ahead, GetAttr): 8 ctx1abcdeghf

374 composed += self.TOKEN_DOT 8 ctx1abcdeghf

375 return composed 8 ctx1abcdeghf

376 

377 def enter(self, subroute: RouteLike) -> Route: 8 ctx1abcdeghf

378 """ 

379 Enter a subroute. 

380 

381 Parameters 

382 ---------- 

383 subroute 

384 A subroute to enter. 

385 

386 """ 

387 return type(self)(self.steps + self.parse(subroute)) 8 ctx1abcdeghf

388 

389 def get(self, obj: Any, /) -> object: 8 ctx1abcdeghf

390 """ 

391 Get an object at the end of this route. 

392 

393 Parameters 

394 ---------- 

395 obj 

396 An object to dive in. 

397 

398 Returns 

399 ------- 

400 The result of visiting the object. 

401 

402 """ 

403 return reduce(lambda obj, step: step(obj), self.__steps, obj) 

404 

405 def set(self, obj: Any, value: object, /) -> None: 8 ctx1abcdeghf

406 """ 

407 Set an object pointed to by this route. 

408 

409 Parameters 

410 ---------- 

411 obj 

412 An object to dive in. 

413 

414 value 

415 A value to set. 

416 

417 Returns 

418 ------- 

419 The result of visiting the object. 

420 

421 """ 

422 steps = self.steps 

423 last_step = steps.pop() 

424 last_step.set( 

425 reduce(lambda obj, step: step(obj), steps, obj), 

426 value, 

427 ) 

428 

429 def __eq__(self, other: object) -> bool: 8 ctx1abcdeghf

430 """ 

431 Compare this route to another route. 

432 

433 Parameters 

434 ---------- 

435 other 

436 Another route to compare to. 

437 

438 """ 

439 if isinstance(other, Route): 439 ↛ 441line 439 didn't jump to line 441, because the condition on line 439 was always true8 ctx1abcdeghf

440 return self.steps == other.steps 8 ctx1abcdeghf

441 if isinstance(other, str): 

442 return self.steps == self.decompose(other) 

443 if isinstance(other, (tuple, list)): 

444 return self.steps == self.parse(other) 

445 return NotImplemented 

446 

447 def __str__(self) -> str: 8 ctx1abcdeghf

448 """Compose this route into a string.""" 

449 return self.compose() 8 ctx1abcdeghf

450 

451 def __iter__(self) -> Iterator[Step[Any]]: 8 ctx1abcdeghf

452 """Yield all steps in this route.""" 

453 yield from self.__steps 8 ctx1abcdeghf

454 

455 def __repr__(self) -> str: 8 ctx1abcdeghf

456 """Represent this route in a string.""" 

457 return f"<{type(self).__name__} {self.compose()!r}>" 8 ctx1abcdeghf

458 

459 

460EMPTY_ROUTE: Route = Route("", allow_empty=True) 8 ctx1abcdeghf

461 

462 

463@class_singledispatch 8 ctx1abcdeghf

464def advance_linked_route( 8 ctx1abcdeghf

465 _current_head: Type[object], # noqa: UP006 

466 _annotation: Any, 

467 _step: Step[object], 

468) -> Any: 

469 """Move one step forward in a linked route.""" 

470 return _AnyHead 8 ctx1abcdeghf

471 

472 

473class _AnyHead: 8 ctx1abcdeghf

474 pass 8 ctx1abcdeghf

475 

476 

477class LinkedRoute(Generic[ConfigObject]): 8 ctx1abcdeghf

478 __head_class: type[Any] 8 ctx1abcdeghf

479 

480 def __init__( 8 ctx1abcdeghf

481 self, 

482 config_class: type[ConfigObject], 

483 route: RouteLike, 

484 ) -> None: 

485 self.__head_class = self.__config_class = config_class 8 ctx1abcdeghf

486 self.__annotation = None 8 ctx1abcdeghf

487 self.__route = EMPTY_ROUTE 8 ctx1abcdeghf

488 for step in Route(route): 8 ctx1abcdeghf

489 self.__step(step) 8 ctx1abcdeghf

490 

491 def __type_check(self, cls: type[object]) -> bool: 8 ctx1abcdeghf

492 # Can't use TypeGuard/TypeIs here, sorry 

493 return isinstance(self.__head_class, type) and issubclass( 8 ctx1abcdeghf

494 self.__head_class, 

495 cls, 

496 ) 

497 

498 @singledispatchmethod 8 ctx1abcdeghf

499 def __step(self, step: Step[Any]) -> None: 8 ctx1abcdeghf

500 raise NotImplementedError 

501 

502 @__step.register 8 ctx1abcdeghf

503 def __getattr(self, step: GetAttr) -> None: 8 ctx1abcdeghf

504 from configzen.config import BaseConfig 8 ctx1abcdeghf

505 

506 head = self.__head_class 8 ctx1abcdeghf

507 

508 if ( 6 ctx1abcdef

509 self.__type_check(BaseConfig) 

510 and step.key not in head.model_fields 

511 and head is not _AnyHead 

512 ): 

513 if head.model_config["extra"] == "allow": 513 ↛ 514line 513 didn't jump to line 514, because the condition on line 513 was never true8 ctx1abcdeghf

514 self.__head_class = _AnyHead 

515 

516 msg = f"Cannot use {step!r} on {self.__head_class.__name__!r}" 8 ctx1abcdeghf

517 raise LinkedRouteError( 8 ctx1abcdeghf

518 msg, 

519 config_class=self.__config_class, 

520 route=str(self.__route), 

521 ) 

522 

523 self.__annotation = annotation = advance_linked_route( 8 ctx1abcdeghf

524 self.__head_class, 

525 self.__annotation, 

526 step, 

527 ) 

528 self.__head_class = get_origin(annotation) or annotation 8 ctx1abcdeghf

529 self.__route = self.__route.enter(step) 8 ctx1abcdeghf

530 

531 @__step.register 8 ctx1abcdeghf

532 def __getitem(self, step: GetItem) -> None: 8 ctx1abcdeghf

533 from configzen.config import BaseConfig 8 ctx1abcdeghf

534 

535 if self.__type_check(BaseConfig): 535 ↛ 536line 535 didn't jump to line 536, because the condition on line 535 was never true8 ctx1abcdeghf

536 msg = f"Cannot use {step!r} on a configuration class" 

537 raise LinkedRouteError( 

538 msg, 

539 config_class=self.__config_class, 

540 route=str(self.__route), 

541 ) 

542 

543 if not hasattr(self.__head_class, "__getitem__"): 543 ↛ 544line 543 didn't jump to line 544, because the condition on line 543 was never true8 ctx1abcdeghf

544 msg = f"Cannot use {step!r} on {self.__head_class.__name__!r}" 

545 raise LinkedRouteError( 

546 msg, 

547 config_class=self.__config_class, 

548 route=str(self.__route), 

549 ) 

550 

551 self.__annotation = annotation = advance_linked_route( 8 ctx1abcdeghf

552 self.__head_class, 

553 self.__annotation, 

554 step, 

555 ) 

556 self.__head_class = get_origin(annotation) or annotation 8 ctx1abcdeghf

557 self.__route = self.__route.enter(step) 8 ctx1abcdeghf

558 

559 def __eq__(self, other: object) -> bool: 8 ctx1abcdeghf

560 if isinstance(other, LinkedRoute): 8 ctx1abcdeghf

561 return ( 8 ctx1abcdeghf

562 self.__config_class is other.__config_class 

563 and self.__route == other.__route 

564 ) 

565 return NotImplemented 

566 

567 def __getitem__(self, item: int | str) -> Self: 8 ctx1abcdeghf

568 self.__step(GetItem(item)) 8 ctx1abcdeghf

569 return self 8 ctx1abcdeghf

570 

571 def __getattr__(self, item: str) -> Self: 8 ctx1abcdeghf

572 self.__step(GetAttr(item)) 8 ctx1abcdeghf

573 return self 8 ctx1abcdeghf

574 

575 def __repr__(self) -> str: 8 ctx1abcdeghf

576 return f"<{type(self).__name__} {self.__route!r}>"