Coverage for configzen/routes.py: 79%
234 statements
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
1"""Routes creation and parsing."""
3from __future__ import annotations 1abcdeghf
5from functools import reduce, singledispatchmethod 1abcdeghf
6from typing import ( 1abcdeghf
7 TYPE_CHECKING,
8 Any,
9 ClassVar,
10 Generic,
11 Type,
12 TypeVar,
13 Union,
14 get_origin,
15)
17from class_singledispatch import class_singledispatch 1abcdeghf
19from configzen.errors import LinkedRouteError, RouteError 1abcdeghf
20from configzen.typedefs import ConfigObject 1abcdeghf
22if TYPE_CHECKING: 1abcdeghf
23 from collections.abc import Iterator
25 from typing_extensions import Self, TypeAlias
28__all__ = ( 1abcdeghf
29 "GetAttr",
30 "GetItem",
31 "Route",
32 "RouteLike",
33 "Step",
34)
36RouteLike: TypeAlias = Union[ 1abcdeghf
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]
47_KT = TypeVar("_KT") 1abcdeghf
50class Step(Generic[_KT]): 1abcdeghf
51 """
52 A configuration route step.
54 Do not use this class directly. Use GetAttr or GetItem instead.
55 """
57 key: _KT 1abcdeghf
59 def __init__(self, key: _KT, /) -> None: 1abcdeghf
60 self.key = key 1abcdeghf
62 def __eq__(self, other: object) -> bool: 1abcdeghf
63 """Compare this step to another step."""
64 if isinstance(other, Step): 1abcdeghf
65 return ( 1abcdeghf
66 issubclass(type(other), type(self))
67 or issubclass(type(self), type(other))
68 ) or self.key == other.key
69 return NotImplemented
71 def get(self, _: Any, /) -> object: 1abcdeghf
72 """Perform a get operation."""
73 raise NotImplementedError
75 def set(self, _: Any, __: object, /) -> None: 1abcdeghf
76 """Perform a set operation."""
77 raise NotImplementedError
79 def __call__(self, obj: Any, /) -> object: 1abcdeghf
80 """Perform a get operation."""
81 return self.get(obj)
83 def __repr__(self) -> str: 1abcdeghf
84 """Represent this step in a string."""
85 return f"{type(self).__name__}({self.key!r})" 1abcdeghf
88class GetAttr(Step[str]): 1abcdeghf
89 """
90 A route step that gets an attribute from an object.
92 The argument is used as an attribute name.
93 """
95 def get(self, target: Any, /) -> object: 1abcdeghf
96 """Get an attribute from an object."""
97 return getattr(target, self.key)
99 def set(self, target: Any, value: object, /) -> None: 1abcdeghf
100 """Set an attribute in an object."""
101 setattr(target, self.key, value)
103 def __str__(self) -> str: 1abcdeghf
104 """Compose this step into a string."""
105 return str(self.key).replace(Route.TOKEN_DOT, r"\.") 1abcdeghf
108class GetItem(Step[Union[int, str]]): 1abcdeghf
109 r"""
110 A route step that gets an item from an object.
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 """
117 def __init__(self, key: int | str, /, *, ignore_digit: bool = False) -> None: 1abcdeghf
118 self.escape = False 1abcdeghf
119 if isinstance(key, str) and key.isdigit(): 1abcdeghf
120 if ignore_digit: 120 ↛ 121line 120 didn't jump to line 121, because the condition on line 120 was never true1abcdeghf
121 self.escape = True
122 else:
123 key = int(key) 1abcdeghf
124 super().__init__(key) 1abcdeghf
126 def get(self, target: Any, /) -> object: 1abcdeghf
127 """Get an item from an object."""
128 return target[self.key]
130 def set(self, target: Any, value: object, /) -> None: 1abcdeghf
131 """Set an item in an object."""
132 target[self.key] = value
134 def __str__(self) -> str: 1abcdeghf
135 """Compose this step into a string."""
136 argument = str(self.key) 1abcdeghf
137 if self.escape: 137 ↛ 138line 137 didn't jump to line 138, because the condition on line 137 was never true1abcdeghf
138 argument = Route.TOKEN_ESCAPE + argument
139 return argument.join( 1abcdeghf
140 Route.TOKEN_ENTER + Route.TOKEN_LEAVE,
141 ).replace(Route.TOKEN_DOT, r"\.")
144def _route_decompose( # noqa: C901, PLR0912, PLR0915 1abcdeghf
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.
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.
168 Returns
169 -------
170 List of steps.
172 """
173 if not route.endswith(dot): 173 ↛ 176line 173 didn't jump to line 176, because the condition on line 173 was always true1abcdeghf
174 route += dot 1abcdeghf
176 key = "" 1abcdeghf
177 entered: int | None = None 1abcdeghf
178 left: bool = False 1abcdeghf
179 steps: list[Step[Any]] = [] 1abcdeghf
180 emit = steps.append 1abcdeghf
181 escaping = False 1abcdeghf
182 escaped = False 1abcdeghf
183 step: Step[Any]
185 for index, token in enumerate(route): 1abcdeghf
186 if escaping: 1abcdeghf
187 key += token 1abcdeghf
188 escaping = False 1abcdeghf
189 escaped = True 1abcdeghf
190 continue 1abcdeghf
191 is_last = index == len(route) - 1 1abcdeghf
192 if token == dot: 1abcdeghf
193 if entered is not None: 193 ↛ 194line 193 didn't jump to line 194, because the condition on line 193 was never true1abcdeghf
194 key += token
195 else:
196 if left: 1abcdeghf
197 step = GetItem(key, ignore_digit=escaped) 1abcdeghf
198 left = False 1abcdeghf
199 else:
200 step = GetAttr(key) 1abcdeghf
201 escaped = False 1abcdeghf
202 emit(step) 1abcdeghf
203 key = "" 1abcdeghf
204 elif token == escape: 1abcdeghf
205 if is_last: 205 ↛ 206line 205 didn't jump to line 206, because the condition on line 205 was never true1abcdeghf
206 key += token
207 else:
208 escaping = True 1abcdeghf
209 elif token == enter: 1abcdeghf
210 if entered is not None: 210 ↛ 211line 210 didn't jump to line 211, because the condition on line 210 was never true1abcdeghf
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 true1abcdeghf
214 entered = index 1abcdeghf
215 if index: 215 ↛ 235line 215 didn't jump to line 235, because the condition on line 215 was always true1abcdeghf
216 if left: 1abcdeghf
217 step = GetItem(key, ignore_digit=escaped) 1abcdeghf
218 left = False 1abcdeghf
219 else:
220 step = GetAttr(key) 1abcdeghf
221 escaped = False 1abcdeghf
222 emit(step) 1abcdeghf
223 key = "" 1abcdeghf
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: 1abcdeghf
228 if entered is None: 228 ↛ 229line 228 didn't jump to line 229, because the condition on line 228 was never true1abcdeghf
229 msg = f"{token!r} not preceded by {enter!r} token"
230 raise RouteError(msg, route=route, index=index)
231 entered = None 1abcdeghf
232 left = True 1abcdeghf
233 else:
234 key += token 1abcdeghf
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 true1abcdeghf
236 msg = f"Expected {leave!r} token"
237 raise RouteError(msg, route=route, index=entered)
238 return steps 1abcdeghf
241class Route: 1abcdeghf
242 r"""
243 Routes are, lists of steps that are used to access values in a configuration.
245 Each step is either a key or an index.
247 A route can be created from a string, a list of steps, or another route.
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')]
257 Parameters
258 ----------
259 route
260 A route to parse.
261 allow_empty
262 Whether to allow empty routes.
264 """
266 TOKEN_DOT: ClassVar[str] = "." 1abcdeghf
267 TOKEN_ESCAPE: ClassVar[str] = "\\" 1abcdeghf
268 TOKEN_ENTER: ClassVar[str] = "[" 1abcdeghf
269 TOKEN_LEAVE: ClassVar[str] = "]" 1abcdeghf
271 TOKENS: ClassVar[tuple[str, str, str, str]] = ( 1abcdeghf
272 TOKEN_DOT,
273 TOKEN_ESCAPE,
274 TOKEN_ENTER,
275 TOKEN_LEAVE,
276 )
278 def __init__( 1abcdeghf
279 self,
280 route: RouteLike,
281 *,
282 allow_empty: bool = False,
283 ) -> None:
284 steps = self.parse(route) 1abcdeghf
285 if not (allow_empty or steps): 285 ↛ 286line 285 didn't jump to line 286, because the condition on line 285 was never true1abcdeghf
286 msg = "Empty configuration route"
287 raise ValueError(msg)
288 self.__steps = tuple(steps) 1abcdeghf
290 @property 1abcdeghf
291 def steps(self) -> list[Step[Any]]: 1abcdeghf
292 """Get all steps in this route."""
293 return list(self.__steps) 1abcdeghf
295 def __hash__(self) -> int: 1abcdeghf
296 """Get a hash of this route."""
297 return hash(self.__steps)
299 @classmethod 1abcdeghf
300 def parse(cls, route: RouteLike) -> list[Step[Any]]: 1abcdeghf
301 """
302 Parse a route into steps.
304 Parameters
305 ----------
306 route
307 The route to parse.
309 Returns
310 -------
311 List of steps.
313 """
314 if isinstance(route, Step): 1abcdeghf
315 return [route] 1abcdeghf
316 if isinstance(route, Route): 316 ↛ 317line 316 didn't jump to line 317, because the condition on line 316 was never true1abcdeghf
317 return route.steps
318 if isinstance(route, (tuple, list)): 1abcdeghf
319 patched_route: list[Step[Any]] = [] 1abcdeghf
320 for element in route: 1abcdeghf
321 if isinstance(element, (str, int)): 321 ↛ 322line 321 didn't jump to line 322, because the condition on line 321 was never true1abcdeghf
322 try:
323 patched_element = next(iter(cls.parse(element)))
324 except StopIteration:
325 continue
326 else:
327 patched_element = element 1abcdeghf
328 patched_route.append(patched_element) 1abcdeghf
329 return patched_route 1abcdeghf
330 if isinstance(route, int): 330 ↛ 331line 330 didn't jump to line 331, because the condition on line 330 was never true1abcdeghf
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 true1abcdeghf
333 return cls.decompose(route) 1abcdeghf
334 msg = f"Invalid route type {type(route)!r}"
335 raise TypeError(msg)
337 @classmethod 1abcdeghf
338 def decompose(cls, route: str) -> list[Step[Any]]: 1abcdeghf
339 """
340 Decompose a route into a list of steps.
342 Parameters
343 ----------
344 route
345 A route to decompose.
347 Returns
348 -------
349 List of steps.
351 """
352 if not route: 1abcdeghf
353 return [] 1abcdeghf
355 dot, escape, enter, leave = cls.TOKENS 1abcdeghf
357 return _route_decompose( 1abcdeghf
358 route,
359 dot=dot,
360 escape=escape,
361 enter=enter,
362 leave=leave,
363 )
365 def compose(self) -> str: 1abcdeghf
366 """Compose this route into a string."""
367 composed = "" 1abcdeghf
368 steps = self.__steps 1abcdeghf
369 for index, step in enumerate(steps): 1abcdeghf
370 composed += str(step) 1abcdeghf
371 if index < len(steps) - 1: 1abcdeghf
372 ahead = steps[index + 1] 1abcdeghf
373 if isinstance(ahead, GetAttr): 1abcdeghf
374 composed += self.TOKEN_DOT 1abcdeghf
375 return composed 1abcdeghf
377 def enter(self, subroute: RouteLike) -> Route: 1abcdeghf
378 """
379 Enter a subroute.
381 Parameters
382 ----------
383 subroute
384 A subroute to enter.
386 """
387 return type(self)(self.steps + self.parse(subroute)) 1abcdeghf
389 def get(self, obj: Any, /) -> object: 1abcdeghf
390 """
391 Get an object at the end of this route.
393 Parameters
394 ----------
395 obj
396 An object to dive in.
398 Returns
399 -------
400 The result of visiting the object.
402 """
403 return reduce(lambda obj, step: step(obj), self.__steps, obj)
405 def set(self, obj: Any, value: object, /) -> None: 1abcdeghf
406 """
407 Set an object pointed to by this route.
409 Parameters
410 ----------
411 obj
412 An object to dive in.
414 value
415 A value to set.
417 Returns
418 -------
419 The result of visiting the object.
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 )
429 def __eq__(self, other: object) -> bool: 1abcdeghf
430 """
431 Compare this route to another route.
433 Parameters
434 ----------
435 other
436 Another route to compare to.
438 """
439 if isinstance(other, Route): 439 ↛ 441line 439 didn't jump to line 441, because the condition on line 439 was always true1abcdeghf
440 return self.steps == other.steps 1abcdeghf
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
447 def __str__(self) -> str: 1abcdeghf
448 """Compose this route into a string."""
449 return self.compose() 1abcdeghf
451 def __iter__(self) -> Iterator[Step[Any]]: 1abcdeghf
452 """Yield all steps in this route."""
453 yield from self.__steps 1abcdeghf
455 def __repr__(self) -> str: 1abcdeghf
456 """Represent this route in a string."""
457 return f"<{type(self).__name__} {self.compose()!r}>" 1abcdeghf
460EMPTY_ROUTE: Route = Route("", allow_empty=True) 1abcdeghf
463@class_singledispatch 1abcdeghf
464def advance_linked_route( 1abcdeghf
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 1abcdeghf
473class _AnyHead: 1abcdeghf
474 pass 1abcdeghf
477class LinkedRoute(Generic[ConfigObject]): 1abcdeghf
478 __head_class: type[Any] 1abcdeghf
480 def __init__( 1abcdeghf
481 self,
482 config_class: type[ConfigObject],
483 route: RouteLike,
484 ) -> None:
485 self.__head_class = self.__config_class = config_class 1abcdeghf
486 self.__annotation = None 1abcdeghf
487 self.__route = EMPTY_ROUTE 1abcdeghf
488 for step in Route(route): 1abcdeghf
489 self.__step(step) 1abcdeghf
491 def __type_check(self, cls: type[object]) -> bool: 1abcdeghf
492 # Can't use TypeGuard/TypeIs here, sorry
493 return isinstance(self.__head_class, type) and issubclass( 1abcdeghf
494 self.__head_class,
495 cls,
496 )
498 @singledispatchmethod 1abcdeghf
499 def __step(self, step: Step[Any]) -> None: 1abcdeghf
500 raise NotImplementedError
502 @__step.register 1abcdeghf
503 def __getattr(self, step: GetAttr) -> None: 1abcdeghf
504 from configzen.config import BaseConfig 1abcdeghf
506 head = self.__head_class 1abcdeghf
508 if ( 1abcdef
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 true1abcdeghf
514 self.__head_class = _AnyHead
516 msg = f"Cannot use {step!r} on {self.__head_class.__name__!r}" 1abcdeghf
517 raise LinkedRouteError( 1abcdeghf
518 msg,
519 config_class=self.__config_class,
520 route=str(self.__route),
521 )
523 self.__annotation = annotation = advance_linked_route( 1abcdeghf
524 self.__head_class,
525 self.__annotation,
526 step,
527 )
528 self.__head_class = get_origin(annotation) or annotation 1abcdeghf
529 self.__route = self.__route.enter(step) 1abcdeghf
531 @__step.register 1abcdeghf
532 def __getitem(self, step: GetItem) -> None: 1abcdeghf
533 from configzen.config import BaseConfig 1abcdeghf
535 if self.__type_check(BaseConfig): 535 ↛ 536line 535 didn't jump to line 536, because the condition on line 535 was never true1abcdeghf
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 )
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 true1abcdeghf
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 )
551 self.__annotation = annotation = advance_linked_route( 1abcdeghf
552 self.__head_class,
553 self.__annotation,
554 step,
555 )
556 self.__head_class = get_origin(annotation) or annotation 1abcdeghf
557 self.__route = self.__route.enter(step) 1abcdeghf
559 def __eq__(self, other: object) -> bool: 1abcdeghf
560 if isinstance(other, LinkedRoute): 1abcdeghf
561 return ( 1abcdeghf
562 self.__config_class is other.__config_class
563 and self.__route == other.__route
564 )
565 return NotImplemented
567 def __getitem__(self, item: int | str) -> Self: 1abcdeghf
568 self.__step(GetItem(item)) 1abcdeghf
569 return self 1abcdeghf
571 def __getattr__(self, item: str) -> Self: 1abcdeghf
572 self.__step(GetAttr(item)) 1abcdeghf
573 return self 1abcdeghf
575 def __repr__(self) -> str: 1abcdeghf
576 return f"<{type(self).__name__} {self.__route!r}>"