Coverage for src/debputy/manifest_parser/declarative_parser.py: 74%
802 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-09-05 06:54 +0000
1import collections
2import dataclasses
3import enum
4import types
5import typing
6from types import UnionType
7from typing import (
8 Any,
9 TypedDict,
10 get_type_hints,
11 Annotated,
12 get_args,
13 get_origin,
14 TypeVar,
15 Generic,
16 Optional,
17 cast,
18 Type,
19 Union,
20 List,
21 NotRequired,
22 Literal,
23 TYPE_CHECKING,
24)
25from collections.abc import Callable, Mapping, Collection, Iterable, Sequence, Container
28from debputy.manifest_parser.base_types import FileSystemMatchRule
29from debputy.manifest_parser.exceptions import (
30 ManifestParseException,
31)
32from debputy.manifest_parser.mapper_code import (
33 normalize_into_list,
34 wrap_into_list,
35 map_each_element,
36)
37from debputy.manifest_parser.parse_hints import (
38 ConditionalRequired,
39 DebputyParseHint,
40 TargetAttribute,
41 ManifestAttribute,
42 ConflictWithSourceAttribute,
43 NotPathHint,
44)
45from debputy.manifest_parser.parser_data import ParserContextData
46from debputy.manifest_parser.tagging_types import (
47 DebputyParsedContent,
48 DebputyDispatchableType,
49 TypeMapping,
50)
51from debputy.manifest_parser.util import (
52 AttributePath,
53 unpack_type,
54 find_annotation,
55 check_integration_mode,
56)
57from debputy.plugin.api.impl_types import (
58 DeclarativeInputParser,
59 TD,
60 ListWrappedDeclarativeInputParser,
61 DispatchingObjectParser,
62 DispatchingTableParser,
63 TTP,
64 TP,
65 InPackageContextParser,
66 AllowNoneDeclarativeInputParser,
67)
68from debputy.plugin.api.spec import (
69 ParserDocumentation,
70 DebputyIntegrationMode,
71 StandardParserAttributeDocumentation,
72 undocumented_attr,
73 ParserAttributeDocumentation,
74 reference_documentation,
75)
76from debputy.util import _info, _warn, assume_not_none
78if TYPE_CHECKING:
79 from debputy.lsp.diagnostics import LintSeverity
82try:
83 from Levenshtein import distance
85 _WARN_ONCE: bool | None = None
86except ImportError:
87 _WARN_ONCE = False
90def _detect_possible_typo(
91 key: str,
92 value: object,
93 manifest_attributes: Mapping[str, "AttributeDescription"],
94 path: "AttributePath",
95) -> None:
96 global _WARN_ONCE
97 if _WARN_ONCE == False:
98 _WARN_ONCE = True
99 _info(
100 "Install python3-levenshtein to have debputy try to detect typos in the manifest."
101 )
102 elif _WARN_ONCE is None:
103 k_len = len(key)
104 key_path = path[key]
105 matches: list[str] = []
106 current_match_strength = 0
107 for acceptable_key, attr in manifest_attributes.items():
108 if abs(k_len - len(acceptable_key)) > 2:
109 continue
110 d = distance(key, acceptable_key)
111 if d > 2:
112 continue
113 try:
114 attr.type_validator.ensure_type(value, key_path)
115 except ManifestParseException:
116 if attr.type_validator.base_type_match(value):
117 match_strength = 1
118 else:
119 match_strength = 0
120 else:
121 match_strength = 2
123 if match_strength < current_match_strength:
124 continue
125 if match_strength > current_match_strength:
126 current_match_strength = match_strength
127 matches.clear()
128 matches.append(acceptable_key)
130 if not matches:
131 return
132 ref = f'at "{path.path}"' if path else "at the manifest root level"
133 if len(matches) == 1:
134 possible_match = repr(matches[0])
135 _warn(
136 f'Possible typo: The key "{key}" {ref} should probably have been {possible_match}'
137 )
138 else:
139 matches.sort()
140 possible_matches = ", ".join(repr(a) for a in matches)
141 _warn(
142 f'Possible typo: The key "{key}" {ref} should probably have been one of {possible_matches}'
143 )
146SF = TypeVar("SF")
147T = TypeVar("T")
148S = TypeVar("S")
151_NONE_TYPE = type(None)
154# These must be able to appear in an "isinstance" check and must be builtin types.
155BASIC_SIMPLE_TYPES = {
156 str: "string",
157 int: "integer",
158 bool: "boolean",
159}
162class AttributeTypeHandler:
163 __slots__ = ("_description", "_ensure_type", "base_type", "mapper")
165 def __init__(
166 self,
167 description: str,
168 ensure_type: Callable[[Any, AttributePath], None],
169 *,
170 base_type: type[Any] | None = None,
171 mapper: (
172 Callable[[Any, AttributePath, Optional["ParserContextData"]], Any]
173 ) | None = None,
174 ) -> None:
175 self._description = description
176 self._ensure_type = ensure_type
177 self.base_type = base_type
178 self.mapper = mapper
180 def describe_type(self) -> str:
181 return self._description
183 def ensure_type(self, obj: object, path: AttributePath) -> None:
184 self._ensure_type(obj, path)
186 def base_type_match(self, obj: object) -> bool:
187 base_type = self.base_type
188 return base_type is not None and isinstance(obj, base_type)
190 def map_type(
191 self,
192 value: Any,
193 path: AttributePath,
194 parser_context: Optional["ParserContextData"],
195 ) -> Any:
196 mapper = self.mapper
197 if mapper is not None:
198 return mapper(value, path, parser_context)
199 return value
201 def combine_mapper(
202 self,
203 mapper: None | (
204 Callable[[Any, AttributePath, Optional["ParserContextData"]], Any]
205 ),
206 ) -> "AttributeTypeHandler":
207 if mapper is None:
208 return self
209 _combined_mapper: Callable[
210 [Any, AttributePath, Optional["ParserContextData"]], Any
211 ]
212 if self.mapper is not None:
213 m = self.mapper
215 def _combined_mapper(
216 value: Any,
217 path: AttributePath,
218 parser_context: Optional["ParserContextData"],
219 ) -> Any:
220 return mapper(m(value, path, parser_context), path, parser_context)
222 else:
223 _combined_mapper = mapper
225 return AttributeTypeHandler(
226 self._description,
227 self._ensure_type,
228 base_type=self.base_type,
229 mapper=_combined_mapper,
230 )
233@dataclasses.dataclass(slots=True)
234class AttributeDescription:
235 source_attribute_name: str
236 target_attribute: str
237 attribute_type: Any
238 type_validator: AttributeTypeHandler
239 annotations: tuple[Any, ...]
240 conflicting_attributes: frozenset[str]
241 conditional_required: Optional["ConditionalRequired"]
242 parse_hints: Optional["DetectedDebputyParseHint"] = None
243 is_optional: bool = False
246def _extract_path_hint(v: Any, attribute_path: AttributePath) -> bool:
247 if attribute_path.path_hint is not None: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 return True
249 if isinstance(v, str):
250 attribute_path.path_hint = v
251 return True
252 elif isinstance(v, list) and len(v) > 0 and isinstance(v[0], str):
253 attribute_path.path_hint = v[0]
254 return True
255 return False
258@dataclasses.dataclass(slots=True, frozen=True)
259class DeclarativeNonMappingInputParser(DeclarativeInputParser[TD], Generic[TD, SF]):
260 alt_form_parser: AttributeDescription
261 inline_reference_documentation: ParserDocumentation | None = None
262 expected_debputy_integration_mode: Container[DebputyIntegrationMode] | None = None
264 def parse_input(
265 self,
266 value: object,
267 path: AttributePath,
268 *,
269 parser_context: Optional["ParserContextData"] = None,
270 ) -> TD:
271 check_integration_mode(
272 path,
273 parser_context,
274 self.expected_debputy_integration_mode,
275 )
276 if self.reference_documentation_url is not None:
277 doc_ref = f" (Documentation: {self.reference_documentation_url})"
278 else:
279 doc_ref = ""
281 alt_form_parser = self.alt_form_parser
282 if value is None: 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 form_note = f" The value must have type: {alt_form_parser.type_validator.describe_type()}"
284 if self.reference_documentation_url is not None:
285 doc_ref = f" Please see {self.reference_documentation_url} for the documentation."
286 raise ManifestParseException(
287 f"The attribute {path.path} was missing a value. {form_note}{doc_ref}"
288 )
289 _extract_path_hint(value, path)
290 alt_form_parser.type_validator.ensure_type(value, path)
291 attribute = alt_form_parser.target_attribute
292 alias_mapping = {
293 attribute: ("", None),
294 }
295 v = alt_form_parser.type_validator.map_type(value, path, parser_context)
296 path.alias_mapping = alias_mapping
297 return cast("TD", {attribute: v})
300@dataclasses.dataclass(slots=True)
301class DeclarativeMappingInputParser(DeclarativeInputParser[TD], Generic[TD, SF]):
302 input_time_required_parameters: frozenset[str]
303 all_parameters: frozenset[str]
304 manifest_attributes: Mapping[str, "AttributeDescription"]
305 source_attributes: Mapping[str, "AttributeDescription"]
306 source_form_attr2source_attributes: Mapping[str, str]
307 at_least_one_of: frozenset[frozenset[str]]
308 alt_form_parser: AttributeDescription | None
309 mutually_exclusive_attributes: frozenset[frozenset[str]] = frozenset()
310 _per_attribute_conflicts_cache: Mapping[str, frozenset[str]] | None = None
311 inline_reference_documentation: ParserDocumentation | None = None
312 path_hint_source_attributes: Sequence[str] = ()
313 expected_debputy_integration_mode: Container[DebputyIntegrationMode] | None = None
315 def _parse_alt_form(
316 self,
317 value: object,
318 path: AttributePath,
319 *,
320 parser_context: Optional["ParserContextData"] = None,
321 ) -> TD:
322 alt_form_parser = self.alt_form_parser
323 if alt_form_parser is None: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 raise ManifestParseException(
325 f"The attribute {path.path} must be a mapping.{self._doc_url_error_suffix()}"
326 )
327 _extract_path_hint(value, path)
328 alt_form_parser.type_validator.ensure_type(value, path)
329 assert (
330 value is not None
331 ), "The alternative form was None, but the parser should have rejected None earlier."
332 attribute = alt_form_parser.target_attribute
333 alias_mapping = {
334 attribute: ("", None),
335 }
336 v = alt_form_parser.type_validator.map_type(value, path, parser_context)
337 path.alias_mapping = alias_mapping
338 return cast("TD", {attribute: v})
340 def _validate_expected_keys(
341 self,
342 value: dict[Any, Any],
343 path: AttributePath,
344 *,
345 parser_context: Optional["ParserContextData"] = None,
346 ) -> None:
347 unknown_keys = value.keys() - self.all_parameters
348 doc_ref = self._doc_url_error_suffix()
349 if unknown_keys: 349 ↛ 350line 349 didn't jump to line 350 because the condition on line 349 was never true
350 for k in unknown_keys:
351 if isinstance(k, str):
352 _detect_possible_typo(k, value[k], self.manifest_attributes, path)
353 unused_keys = self.all_parameters - value.keys()
354 if unused_keys:
355 k = ", ".join(unused_keys)
356 raise ManifestParseException(
357 f'Unknown keys "{unknown_keys}" at {path.path_container_lc}". Keys that could be used here are: {k}.{doc_ref}'
358 )
359 raise ManifestParseException(
360 f'Unknown keys "{unknown_keys}" at {path.path_container_lc}". Please remove them.{doc_ref}'
361 )
362 missing_keys = self.input_time_required_parameters - value.keys()
363 if missing_keys:
364 required = ", ".join(repr(k) for k in sorted(missing_keys))
365 raise ManifestParseException(
366 f"The following keys were required but not present at {path.path_container_lc}: {required}{doc_ref}"
367 )
368 for maybe_required in self.all_parameters - value.keys():
369 attr = self.manifest_attributes[maybe_required]
370 assert attr.conditional_required is None or parser_context is not None
371 if ( 371 ↛ 377line 371 didn't jump to line 377 because the condition on line 371 was never true
372 attr.conditional_required is not None
373 and attr.conditional_required.condition_applies(
374 assume_not_none(parser_context)
375 )
376 ):
377 reason = attr.conditional_required.reason
378 raise ManifestParseException(
379 f'Missing the *conditionally* required attribute "{maybe_required}" at {path.path_container_lc}. {reason}{doc_ref}'
380 )
381 for keyset in self.at_least_one_of:
382 matched_keys = value.keys() & keyset
383 if not matched_keys: 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true
384 conditionally_required = ", ".join(repr(k) for k in sorted(keyset))
385 raise ManifestParseException(
386 f"At least one of the following keys must be present at {path.path_container_lc}:"
387 f" {conditionally_required}{doc_ref}"
388 )
389 for group in self.mutually_exclusive_attributes:
390 matched = value.keys() & group
391 if len(matched) > 1: 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 ck = ", ".join(repr(k) for k in sorted(matched))
393 raise ManifestParseException(
394 f"Could not parse {path.path_container_lc}: The following attributes are"
395 f" mutually exclusive: {ck}{doc_ref}"
396 )
398 def _parse_typed_dict_form(
399 self,
400 value: dict[Any, Any],
401 path: AttributePath,
402 *,
403 parser_context: Optional["ParserContextData"] = None,
404 ) -> TD:
405 self._validate_expected_keys(value, path, parser_context=parser_context)
406 result = {}
407 per_attribute_conflicts = self._per_attribute_conflicts()
408 alias_mapping = {}
409 for path_hint_source_attributes in self.path_hint_source_attributes:
410 v = value.get(path_hint_source_attributes)
411 if v is not None and _extract_path_hint(v, path):
412 break
413 for k, v in value.items():
414 attr = self.manifest_attributes[k]
415 matched = value.keys() & per_attribute_conflicts[k]
416 if matched: 416 ↛ 417line 416 didn't jump to line 417 because the condition on line 416 was never true
417 ck = ", ".join(repr(k) for k in sorted(matched))
418 raise ManifestParseException(
419 f'The attribute "{k}" at {path.path} cannot be used with the following'
420 f" attributes: {ck}{self._doc_url_error_suffix()}"
421 )
422 nk = attr.target_attribute
423 key_path = path[k]
424 attr.type_validator.ensure_type(v, key_path)
425 if v is None: 425 ↛ 426line 425 didn't jump to line 426 because the condition on line 425 was never true
426 continue
427 if k != nk:
428 alias_mapping[nk] = k, None
429 v = attr.type_validator.map_type(v, key_path, parser_context)
430 result[nk] = v
431 if alias_mapping:
432 path.alias_mapping = alias_mapping
433 return cast("TD", result)
435 def _doc_url_error_suffix(self, *, see_url_version: bool = False) -> str:
436 doc_url = self.reference_documentation_url
437 if doc_url is not None:
438 if see_url_version: 438 ↛ 439line 438 didn't jump to line 439 because the condition on line 438 was never true
439 return f" Please see {doc_url} for the documentation."
440 return f" (Documentation: {doc_url})"
441 return ""
443 def parse_input(
444 self,
445 value: object,
446 path: AttributePath,
447 *,
448 parser_context: Optional["ParserContextData"] = None,
449 ) -> TD:
450 check_integration_mode(
451 path,
452 parser_context,
453 self.expected_debputy_integration_mode,
454 )
455 if value is None: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 form_note = " The attribute must be a mapping."
457 if self.alt_form_parser is not None:
458 form_note = (
459 " The attribute can be a mapping or a non-mapping format"
460 ' (usually, "non-mapping format" means a string or a list of strings).'
461 )
462 doc_ref = self._doc_url_error_suffix(see_url_version=True)
463 raise ManifestParseException(
464 f"The attribute {path.path} was missing a value. {form_note}{doc_ref}"
465 )
467 if not isinstance(value, dict):
468 return self._parse_alt_form(value, path, parser_context=parser_context)
469 return self._parse_typed_dict_form(value, path, parser_context=parser_context)
471 def _per_attribute_conflicts(self) -> Mapping[str, frozenset[str]]:
472 conflicts = self._per_attribute_conflicts_cache
473 if conflicts is not None:
474 return conflicts
475 attrs = self.source_attributes
476 conflicts = {
477 a.source_attribute_name: frozenset(
478 attrs[ca].source_attribute_name for ca in a.conflicting_attributes
479 )
480 for a in attrs.values()
481 }
482 self._per_attribute_conflicts_cache = conflicts
483 return conflicts
486def _is_path_attribute_candidate(
487 source_attribute: AttributeDescription, target_attribute: AttributeDescription
488) -> bool:
489 if (
490 source_attribute.parse_hints
491 and not source_attribute.parse_hints.applicable_as_path_hint
492 ):
493 return False
494 target_type = target_attribute.attribute_type
495 _, origin, args = unpack_type(target_type, False)
496 match_type = target_type
497 if origin == list:
498 match_type = args[0]
499 return isinstance(match_type, type) and issubclass(match_type, FileSystemMatchRule)
502def is_typeddict(t: Any) -> bool:
503 return typing.is_typeddict(t) or (
504 # Logically, not is_typeddict(t) and is subclass(DebputyParsedContent)
505 # implies not is_typeddict(DebputyParsedContent)
506 # except that subclass *fails* for typeddicts.
507 not typing.is_typeddict(DebputyParsedContent)
508 and isinstance(t, type)
509 and issubclass(t, DebputyParsedContent)
510 )
513class ParserGenerator:
514 def __init__(self) -> None:
515 self._registered_types: dict[Any, TypeMapping[Any, Any]] = {}
516 self._object_parsers: dict[str, DispatchingObjectParser] = {}
517 self._table_parsers: dict[
518 type[DebputyDispatchableType], DispatchingTableParser[Any]
519 ] = {}
520 self._in_package_context_parser: dict[str, Any] = {}
522 def register_mapped_type(self, mapped_type: TypeMapping[Any, Any]) -> None:
523 existing = self._registered_types.get(mapped_type.target_type)
524 if existing is not None: 524 ↛ 525line 524 didn't jump to line 525 because the condition on line 524 was never true
525 raise ValueError(f"The type {existing} is already registered")
526 self._registered_types[mapped_type.target_type] = mapped_type
528 def get_mapped_type_from_target_type(
529 self,
530 mapped_type: type[T],
531 ) -> TypeMapping[Any, T] | None:
532 return self._registered_types.get(mapped_type)
534 def discard_mapped_type(self, mapped_type: type[T]) -> None:
535 del self._registered_types[mapped_type]
537 def add_table_parser(self, rt: type[DebputyDispatchableType], path: str) -> None:
538 assert rt not in self._table_parsers
539 self._table_parsers[rt] = DispatchingTableParser(rt, path)
541 def add_object_parser(
542 self,
543 path: str,
544 *,
545 parser_documentation: ParserDocumentation | None = None,
546 expected_debputy_integration_mode: None | (
547 Container[DebputyIntegrationMode]
548 ) = None,
549 unknown_keys_diagnostic_severity: Optional["LintSeverity"] = "error",
550 allow_unknown_keys: bool = False,
551 ) -> DispatchingObjectParser:
552 assert path not in self._in_package_context_parser
553 assert path not in self._object_parsers
554 object_parser = DispatchingObjectParser(
555 path,
556 parser_documentation=parser_documentation,
557 expected_debputy_integration_mode=expected_debputy_integration_mode,
558 unknown_keys_diagnostic_severity=unknown_keys_diagnostic_severity,
559 allow_unknown_keys=allow_unknown_keys,
560 )
561 self._object_parsers[path] = object_parser
562 return object_parser
564 def add_in_package_context_parser(
565 self,
566 path: str,
567 delegate: DeclarativeInputParser[Any],
568 ) -> None:
569 assert path not in self._in_package_context_parser
570 assert path not in self._object_parsers
571 self._in_package_context_parser[path] = InPackageContextParser(path, delegate)
573 @property
574 def dispatchable_table_parsers(
575 self,
576 ) -> Mapping[type[DebputyDispatchableType], DispatchingTableParser[Any]]:
577 return self._table_parsers
579 @property
580 def dispatchable_object_parsers(self) -> Mapping[str, DispatchingObjectParser]:
581 return self._object_parsers
583 def dispatch_parser_table_for(
584 self, rule_type: TTP
585 ) -> DispatchingTableParser[TP] | None:
586 return cast(
587 "Optional[DispatchingTableParser[TP]]", self._table_parsers.get(rule_type)
588 )
590 def generate_parser(
591 self,
592 parsed_content: type[TD],
593 *,
594 source_content: SF | None = None,
595 allow_none_value: bool = False,
596 allow_none_attributes: bool = False,
597 inline_reference_documentation: ParserDocumentation | None = None,
598 expected_debputy_integration_mode: None | (
599 Container[DebputyIntegrationMode]
600 ) = None,
601 automatic_docs: None | (
602 Mapping[type[Any], Sequence[StandardParserAttributeDocumentation]]
603 ) = None,
604 ) -> DeclarativeInputParser[TD]:
605 """Derive a parser from a TypedDict
607 Generates a parser for a segment of the manifest (think the `install-docs` snippet) from a TypedDict
608 or two that are used as a description.
610 In its most simple use-case, the caller provides a TypedDict of the expected attributed along with
611 their types. As an example:
613 >>> class InstallDocsRule(DebputyParsedContent):
614 ... sources: list[str]
615 ... into: list[str] # TODO, bad example (but better example requires set up)
616 >>> pg = ParserGenerator()
617 >>> simple_parser = pg.generate_parser(InstallDocsRule)
619 This will create a parser that would be able to interpret something like:
621 ```yaml
622 install-docs:
623 sources: ["docs/*"]
624 into: ["my-pkg"]
625 ```
627 While this is sufficient for programmers, it is a bit rigid for the packager writing the manifest. Therefore,
628 you can also provide a TypedDict describing the input, enabling more flexibility:
630 >>> class InstallDocsRule(DebputyParsedContent):
631 ... sources: list[str]
632 ... into: list[str] # TODO, bad example (but better example requires set up)
633 >>> class InputDocsRuleInputFormat(TypedDict):
634 ... source: NotRequired[Annotated[str, DebputyParseHint.target_attribute("sources")]]
635 ... sources: NotRequired[list[str]]
636 ... into: str | list[str] # TODO, bad example (but better example requires set up)
637 >>> pg = ParserGenerator()
638 >>> flexible_parser = pg.generate_parser(
639 ... InstallDocsRule,
640 ... source_content=InputDocsRuleInputFormat,
641 ... )
643 In this case, the `sources` field can either come from a single `source` in the manifest (which must be a string)
644 or `sources` (which must be a list of strings). The parser also ensures that only one of `source` or `sources`
645 is used to ensure the input is not ambiguous. For the `into` parameter, the parser will accept it being a str
646 or a list of strings. Regardless of how the input was provided, the parser will normalize the input so that
647 both `sources` and `into` in the result is a list of strings. As an example, this parser can accept
648 both the previous input but also the following input:
650 ```yaml
651 install-docs:
652 source: "docs/*"
653 into: "my-pkg"
654 ```
656 The `source` and `into` attributes are then normalized to lists as if the user had written them as lists
657 with a single string in them. As noted above, the name of the `source` attribute will also be normalized
658 while parsing.
660 In the cases where only one field is required by the user, it can sometimes make sense to allow a non-dict
661 as part of the input. Example:
663 >>> class DiscardRule(DebputyParsedContent):
664 ... paths: List[str]
665 >>> class DiscardRuleInputDictFormat(TypedDict):
666 ... path: NotRequired[Annotated[str, DebputyParseHint.target_attribute("paths")]]
667 ... paths: NotRequired[List[str]]
668 >>> # This format relies on DiscardRule having exactly one Required attribute
669 >>> DiscardRuleInputWithAltFormat = Union[
670 ... DiscardRuleInputDictFormat,
671 ... str,
672 ... List[str],
673 ... ]
674 >>> pg = ParserGenerator()
675 >>> flexible_parser = pg.generate_parser(
676 ... DiscardRule,
677 ... source_content=DiscardRuleInputWithAltFormat,
678 ... )
681 Supported types:
682 * `List` - must have a fixed type argument (such as `List[str]`)
683 * `str`
684 * `int`
685 * `BinaryPackage` - When provided (or required), the user must provide a package name listed
686 in the debian/control file. The code receives the BinaryPackage instance
687 matching that input.
688 * `FileSystemMode` - When provided (or required), the user must provide a file system mode in any
689 format that `debputy' provides (such as `0644` or `a=rw,go=rw`).
690 * `FileSystemOwner` - When provided (or required), the user must a file system owner that is
691 available statically on all Debian systems (must be in `base-passwd`).
692 The user has multiple options for how to specify it (either via name or id).
693 * `FileSystemGroup` - When provided (or required), the user must a file system group that is
694 available statically on all Debian systems (must be in `base-passwd`).
695 The user has multiple options for how to specify it (either via name or id).
696 * `ManifestCondition` - When provided (or required), the user must specify a conditional rule to apply.
697 Usually, it is better to extend `DebputyParsedContentStandardConditional`, which
698 provides the `debputy' default `when` parameter for conditionals.
700 Supported special type-like parameters:
702 * `Required` / `NotRequired` to mark a field as `Required` or `NotRequired`. Must be provided at the
703 outermost level. Cannot vary between `parsed_content` and `source_content`.
704 * `Annotated`. Accepted at the outermost level (inside Required/NotRequired) but ignored at the moment.
705 * `Union`. Must be the outermost level (inside `Annotated` or/and `Required`/`NotRequired` if these are present).
706 Automapping (see below) is restricted to two members in the Union.
708 Notable non-supported types:
709 * `Mapping` and all variants therefore (such as `dict`). In the future, nested `TypedDict`s may be allowed.
710 * `Optional` (or `Union[..., None]`): Use `NotRequired` for optional fields.
712 Automatic mapping rules from `source_content` to `parsed_content`:
713 - `Union[T, List[T]]` can be narrowed automatically to `List[T]`. Transformation is basically:
714 `lambda value: value if isinstance(value, list) else [value]`
715 - `T` can be mapped automatically to `List[T]`, Transformation being: `lambda value: [value]`
717 Additionally, types can be annotated (`Annotated[str, ...]`) with `DebputyParseHint`s. Check its classmethod
718 for concrete features that may be useful to you.
720 :param parsed_content: A DebputyParsedContent / TypedDict describing the desired model of the input once parsed.
721 (DebputyParsedContent is a TypedDict subclass that work around some inadequate type checkers).
722 It can also be a `List[DebputyParsedContent]`. In that case, `source_content` must be a
723 `List[TypedDict[...]]`.
724 :param source_content: Optionally, a TypedDict describing the input allowed by the user. This can be useful
725 to describe more variations than in `parsed_content` that the parser will normalize for you. If omitted,
726 the parsed_content is also considered the source_content (which affects what annotations are allowed in it).
727 Note you should never pass the parsed_content as source_content directly.
728 :param allow_none_value: Allow `None` as a valid input value, causing the parsed form to be None with no errors.
729 Note that the `parsed_content` (and the `source_content` if given) must *not* be unioned with optional still.
730 :param allow_none_attributes: In rare cases, you want to support explicitly provided vs. optional in the attributes.
731 In this case, you should set this to True. Though, in 99.9% of all cases, you want `NotRequired` rather
732 than `Optional` (and can keep this False).
733 :param inline_reference_documentation: Optionally, programmatic documentation
734 :param expected_debputy_integration_mode: If provided, this declares the integration modes where the
735 result of the parser can be used. This is primarily useful for "fail-fast" on incorrect usage.
736 When the restriction is not satisfiable, the generated parser will trigger a parse error immediately
737 (resulting in a "compile time" failure rather than a "runtime" failure).
738 :return: An input parser capable of reading input matching the TypedDict(s) used as reference.
739 """
740 orig_parsed_content = parsed_content
741 if source_content is parsed_content: 741 ↛ 742line 741 didn't jump to line 742 because the condition on line 741 was never true
742 raise ValueError(
743 "Do not provide source_content if it is the same as parsed_content"
744 )
745 is_list_wrapped = False
746 if get_origin(orig_parsed_content) == list:
747 parsed_content = get_args(orig_parsed_content)[0]
748 is_list_wrapped = True
750 if isinstance(parsed_content, type) and issubclass(
751 parsed_content, DebputyDispatchableType
752 ):
753 parser = self.dispatch_parser_table_for(parsed_content)
754 if parser is None: 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 raise ValueError(
756 f"Unsupported parsed_content descriptor: {parsed_content.__qualname__}."
757 f" The class {parsed_content.__qualname__} is not a pre-registered type."
758 )
759 # Only the list wrapped version has documentation, and we cannot rely on the delegate
760 # for `expected_debputy_integration_mode` (see test_debputy_lint_integration_mode).
761 #
762 # To be fair, the dispatched type does not know the details of where it is being
763 # dispatched from and in theory, it could be used in multiple places with different
764 # integration modes (`ManifestCondition` is used from many places as a half examples).
765 if is_list_wrapped: 765 ↛ 771line 765 didn't jump to line 771 because the condition on line 765 was always true
766 parser = ListWrappedDeclarativeInputParser(
767 parser,
768 inline_reference_documentation=inline_reference_documentation,
769 expected_debputy_integration_mode=expected_debputy_integration_mode,
770 )
771 return parser
773 if not is_typeddict(parsed_content): 773 ↛ 774line 773 didn't jump to line 774 because the condition on line 773 was never true
774 raise ValueError(
775 f"Unsupported parsed_content descriptor: {parsed_content.__qualname__}."
776 ' Only "TypedDict"-based types and a subset of "DebputyDispatchableType" are supported.'
777 )
778 if is_list_wrapped and source_content is not None:
779 if get_origin(source_content) != list: 779 ↛ 780line 779 didn't jump to line 780 because the condition on line 779 was never true
780 raise ValueError(
781 "If the parsed_content is a List type, then source_format must be a List type as well."
782 )
783 source_content = get_args(source_content)[0]
785 target_attributes = self._parse_types(
786 parsed_content,
787 allow_source_attribute_annotations=source_content is None,
788 forbid_optional=not allow_none_attributes,
789 )
790 required_target_parameters = frozenset(parsed_content.__required_keys__)
791 parsed_alt_form = None
792 non_mapping_source_only = False
794 if source_content is not None:
795 default_target_attribute = None
796 if len(required_target_parameters) == 1:
797 default_target_attribute = next(iter(required_target_parameters))
799 source_typed_dict, alt_source_forms = _extract_typed_dict(
800 source_content,
801 default_target_attribute,
802 )
803 if alt_source_forms:
804 parsed_alt_form = self._parse_alt_form(
805 alt_source_forms,
806 default_target_attribute,
807 )
808 if source_typed_dict is not None:
809 source_content_attributes = self._parse_types(
810 source_typed_dict,
811 allow_target_attribute_annotation=True,
812 allow_source_attribute_annotations=True,
813 forbid_optional=not allow_none_attributes,
814 )
815 source_content_parameter = "source_content"
816 source_and_parsed_differs = True
817 else:
818 source_typed_dict = parsed_content
819 source_content_attributes = target_attributes
820 source_content_parameter = "parsed_content"
821 source_and_parsed_differs = True
822 non_mapping_source_only = True
823 else:
824 source_typed_dict = parsed_content
825 source_content_attributes = target_attributes
826 source_content_parameter = "parsed_content"
827 source_and_parsed_differs = False
829 sources = collections.defaultdict(set)
830 seen_targets = set()
831 seen_source_names: dict[str, str] = {}
832 source_attributes: dict[str, AttributeDescription] = {}
833 path_hint_source_attributes = []
835 for k in source_content_attributes:
836 ia = source_content_attributes[k]
838 ta = (
839 target_attributes.get(ia.target_attribute)
840 if source_and_parsed_differs
841 else ia
842 )
843 if ta is None: 843 ↛ 845line 843 didn't jump to line 845 because the condition on line 843 was never true
844 # Error message would be wrong if this assertion is false.
845 assert source_and_parsed_differs
846 raise ValueError(
847 f'The attribute "{k}" from the "source_content" parameter should have mapped'
848 f' to "{ia.target_attribute}", but that parameter does not exist in "parsed_content"'
849 )
850 if _is_path_attribute_candidate(ia, ta):
851 path_hint_source_attributes.append(ia.source_attribute_name)
852 existing_source_name = seen_source_names.get(ia.source_attribute_name)
853 if existing_source_name: 853 ↛ 854line 853 didn't jump to line 854 because the condition on line 853 was never true
854 raise ValueError(
855 f'The attribute "{k}" and "{existing_source_name}" both share the source name'
856 f' "{ia.source_attribute_name}". Please change the {source_content_parameter} parameter,'
857 f' so only one attribute use "{ia.source_attribute_name}".'
858 )
859 seen_source_names[ia.source_attribute_name] = k
860 seen_targets.add(ta.target_attribute)
861 sources[ia.target_attribute].add(k)
862 if source_and_parsed_differs:
863 bridge_mapper = self._type_normalize(
864 k, ia.attribute_type, ta.attribute_type, False
865 )
866 ia.type_validator = ia.type_validator.combine_mapper(bridge_mapper)
867 source_attributes[k] = ia
869 def _as_attr_names(td_name: Iterable[str]) -> frozenset[str]:
870 return frozenset(
871 source_content_attributes[a].source_attribute_name for a in td_name
872 )
874 _check_attributes(
875 parsed_content,
876 source_typed_dict,
877 source_content_attributes,
878 sources,
879 )
881 at_least_one_of = frozenset(
882 _as_attr_names(g)
883 for k, g in sources.items()
884 if len(g) > 1 and k in required_target_parameters
885 )
887 if source_and_parsed_differs and seen_targets != target_attributes.keys(): 887 ↛ 888line 887 didn't jump to line 888 because the condition on line 887 was never true
888 missing = ", ".join(
889 repr(k) for k in (target_attributes.keys() - seen_targets)
890 )
891 raise ValueError(
892 'The following attributes in "parsed_content" did not have a source field in "source_content":'
893 f" {missing}"
894 )
895 all_mutually_exclusive_fields = frozenset(
896 _as_attr_names(g) for g in sources.values() if len(g) > 1
897 )
899 all_parameters = (
900 source_typed_dict.__required_keys__ | source_typed_dict.__optional_keys__
901 )
902 _check_conflicts(
903 source_content_attributes,
904 source_typed_dict.__required_keys__,
905 all_parameters,
906 )
908 manifest_attributes = {
909 a.source_attribute_name: a for a in source_content_attributes.values()
910 }
912 if parsed_alt_form is not None:
913 target_attribute = parsed_alt_form.target_attribute
914 if ( 914 ↛ 919line 914 didn't jump to line 919 because the condition on line 914 was never true
915 target_attribute not in required_target_parameters
916 and required_target_parameters
917 or len(required_target_parameters) > 1
918 ):
919 raise NotImplementedError(
920 "When using alternative source formats (Union[TypedDict, ...]), then the"
921 " target must have at most one require parameter"
922 )
923 bridge_mapper = self._type_normalize(
924 target_attribute,
925 parsed_alt_form.attribute_type,
926 target_attributes[target_attribute].attribute_type,
927 False,
928 )
929 parsed_alt_form.type_validator = (
930 parsed_alt_form.type_validator.combine_mapper(bridge_mapper)
931 )
933 inline_reference_documentation = (
934 _verify_and_auto_correct_inline_reference_documentation(
935 parsed_content,
936 source_typed_dict,
937 source_content_attributes,
938 inline_reference_documentation,
939 parsed_alt_form is not None,
940 automatic_docs,
941 )
942 )
943 if non_mapping_source_only:
944 parser = DeclarativeNonMappingInputParser(
945 assume_not_none(parsed_alt_form),
946 inline_reference_documentation=inline_reference_documentation,
947 expected_debputy_integration_mode=expected_debputy_integration_mode,
948 )
949 else:
950 parser = DeclarativeMappingInputParser(
951 _as_attr_names(source_typed_dict.__required_keys__),
952 _as_attr_names(all_parameters),
953 manifest_attributes,
954 source_attributes,
955 source_form_attr2source_attributes={
956 v.source_attribute_name: sa for (sa, v) in source_attributes.items()
957 },
958 mutually_exclusive_attributes=all_mutually_exclusive_fields,
959 alt_form_parser=parsed_alt_form,
960 at_least_one_of=at_least_one_of,
961 inline_reference_documentation=inline_reference_documentation,
962 path_hint_source_attributes=tuple(path_hint_source_attributes),
963 expected_debputy_integration_mode=expected_debputy_integration_mode,
964 )
965 if is_list_wrapped:
966 parser = ListWrappedDeclarativeInputParser(
967 parser,
968 )
969 if allow_none_value:
970 parser = AllowNoneDeclarativeInputParser(
971 parser,
972 )
973 return parser
975 def _as_type_validator(
976 self,
977 attribute: str,
978 provided_type: Any,
979 parsing_typed_dict_attribute: bool,
980 ) -> AttributeTypeHandler:
981 assert not isinstance(provided_type, tuple)
983 if isinstance(provided_type, type):
984 if issubclass(provided_type, DebputyDispatchableType):
985 return _dispatch_parser(provided_type)
986 if issubclass(provided_type, enum.StrEnum):
987 return _handle_str_enum(provided_type)
989 unmapped_type = self._strip_mapped_types(
990 provided_type,
991 parsing_typed_dict_attribute,
992 )
993 type_normalizer = self._type_normalize(
994 attribute,
995 unmapped_type,
996 provided_type,
997 parsing_typed_dict_attribute,
998 )
999 t_unmapped, t_unmapped_orig, t_unmapped_args = unpack_type(
1000 unmapped_type,
1001 parsing_typed_dict_attribute,
1002 )
1003 _, t_provided_orig, t_provided_args = unpack_type(
1004 provided_type,
1005 parsing_typed_dict_attribute,
1006 )
1008 if (
1009 (t_unmapped_orig == Union or t_unmapped_orig == types.UnionType)
1010 and t_unmapped_args
1011 and len(t_unmapped_args) == 2
1012 and any(v is _NONE_TYPE for v in t_unmapped_args)
1013 ):
1014 _, _, args = unpack_type(provided_type, parsing_typed_dict_attribute)
1015 actual_type = [a for a in args if a is not _NONE_TYPE][0]
1016 validator = self._as_type_validator(
1017 attribute, actual_type, parsing_typed_dict_attribute
1018 )
1020 def _validator(v: Any, path: AttributePath) -> None:
1021 if v is None:
1022 return
1023 validator.ensure_type(v, path)
1025 return AttributeTypeHandler(
1026 validator.describe_type(),
1027 _validator,
1028 base_type=validator.base_type,
1029 mapper=type_normalizer,
1030 )
1032 if unmapped_type in BASIC_SIMPLE_TYPES:
1033 type_name = BASIC_SIMPLE_TYPES[unmapped_type]
1035 type_mapping = self._registered_types.get(provided_type)
1036 if type_mapping is not None:
1037 simple_type = f" ({type_name})"
1038 type_name = type_mapping.target_type.__name__
1039 else:
1040 simple_type = ""
1042 def _validator(v: Any, path: AttributePath) -> None:
1043 if not isinstance(v, unmapped_type):
1044 _validation_type_error(
1045 path, f"The attribute must be a {type_name}{simple_type}"
1046 )
1048 return AttributeTypeHandler(
1049 type_name,
1050 _validator,
1051 base_type=unmapped_type,
1052 mapper=type_normalizer,
1053 )
1054 if t_unmapped_orig == list:
1055 if not t_unmapped_args: 1055 ↛ 1056line 1055 didn't jump to line 1056 because the condition on line 1055 was never true
1056 raise ValueError(
1057 f'The attribute "{attribute}" is List but does not have Generics (Must use List[X])'
1058 )
1060 genetic_type = t_unmapped_args[0]
1061 key_mapper = self._as_type_validator(
1062 attribute,
1063 genetic_type,
1064 parsing_typed_dict_attribute,
1065 )
1067 def _validator(v: Any, path: AttributePath) -> None:
1068 if not isinstance(v, list):
1069 _validation_type_error(path, "The attribute must be a list")
1070 for i, list_item in enumerate(v):
1071 key_mapper.ensure_type(list_item, path[i])
1073 list_mapper = (
1074 map_each_element(key_mapper.mapper)
1075 if key_mapper.mapper is not None
1076 else None
1077 )
1079 return AttributeTypeHandler(
1080 f"List of {key_mapper.describe_type()}",
1081 _validator,
1082 base_type=list,
1083 mapper=type_normalizer,
1084 ).combine_mapper(list_mapper)
1085 if is_typeddict(provided_type):
1086 subparser = self.generate_parser(cast("Type[TD]", provided_type))
1087 return AttributeTypeHandler(
1088 description=f"{provided_type.__name__} (Typed Mapping)",
1089 ensure_type=lambda v, ap: None,
1090 base_type=dict,
1091 mapper=lambda v, ap, cv: subparser.parse_input(
1092 v, ap, parser_context=cv
1093 ),
1094 )
1095 if t_unmapped_orig == dict:
1096 if not t_unmapped_args or len(t_unmapped_args) != 2: 1096 ↛ 1097line 1096 didn't jump to line 1097 because the condition on line 1096 was never true
1097 raise ValueError(
1098 f'The attribute "{attribute}" is Dict but does not have Generics (Must use Dict[str, Y])'
1099 )
1100 if t_unmapped_args[0] != str: 1100 ↛ 1101line 1100 didn't jump to line 1101 because the condition on line 1100 was never true
1101 raise ValueError(
1102 f'The attribute "{attribute}" is Dict and has a non-str type as key.'
1103 " Currently, only `str` is supported (Dict[str, Y])"
1104 )
1105 key_mapper = self._as_type_validator(
1106 attribute,
1107 t_unmapped_args[0],
1108 parsing_typed_dict_attribute,
1109 )
1110 value_mapper = self._as_type_validator(
1111 attribute,
1112 t_unmapped_args[1],
1113 parsing_typed_dict_attribute,
1114 )
1116 if key_mapper.base_type is None: 1116 ↛ 1117line 1116 didn't jump to line 1117 because the condition on line 1116 was never true
1117 raise ValueError(
1118 f'The attribute "{attribute}" is Dict and the key did not have a trivial base type. Key types'
1119 f" without trivial base types (such as `str`) are not supported at the moment."
1120 )
1122 if value_mapper.mapper is not None: 1122 ↛ 1123line 1122 didn't jump to line 1123 because the condition on line 1122 was never true
1123 raise ValueError(
1124 f'The attribute "{attribute}" is Dict and the value requires mapping.'
1125 " Currently, this is not supported. Consider a simpler type (such as Dict[str, str] or Dict[str, Any])."
1126 " Better typing may come later"
1127 )
1129 def _validator(v: Any, path: AttributePath) -> None:
1130 if not isinstance(v, dict): 1130 ↛ 1131line 1130 didn't jump to line 1131 because the condition on line 1130 was never true
1131 _validation_type_error(path, "The attribute must be a mapping")
1132 key_name = "the first key in the mapping"
1133 for i, (k, value) in enumerate(v.items()):
1134 if not key_mapper.base_type_match(k): 1134 ↛ 1135line 1134 didn't jump to line 1135 because the condition on line 1134 was never true
1135 kp = path.copy_with_path_hint(key_name)
1136 _validation_type_error(
1137 kp,
1138 f'The key number {i + 1} in attribute "{kp}" must be a {key_mapper.describe_type()}',
1139 )
1140 key_name = f"the key after {k}"
1141 value_mapper.ensure_type(value, path[k])
1143 return AttributeTypeHandler(
1144 f"Mapping of {value_mapper.describe_type()}",
1145 _validator,
1146 base_type=dict,
1147 mapper=type_normalizer,
1148 ).combine_mapper(key_mapper.mapper)
1149 if t_unmapped_orig in (Union, UnionType):
1150 if _is_two_arg_x_list_x(t_provided_args):
1151 # Force the order to be "X, List[X]" as it simplifies the code
1152 x_list_x = (
1153 t_provided_args
1154 if get_origin(t_provided_args[1]) == list
1155 else (t_provided_args[1], t_provided_args[0])
1156 )
1158 # X, List[X] could match if X was List[Y]. However, our code below assumes
1159 # that X is a non-list. The `_is_two_arg_x_list_x` returns False for this
1160 # case to avoid this assert and fall into the "generic case".
1161 assert get_origin(x_list_x[0]) != list
1162 x_subtype_checker = self._as_type_validator(
1163 attribute,
1164 x_list_x[0],
1165 parsing_typed_dict_attribute,
1166 )
1167 list_x_subtype_checker = self._as_type_validator(
1168 attribute,
1169 x_list_x[1],
1170 parsing_typed_dict_attribute,
1171 )
1172 type_description = x_subtype_checker.describe_type()
1173 type_description = f"{type_description} or a list of {type_description}"
1175 def _validator(v: Any, path: AttributePath) -> None:
1176 if isinstance(v, list):
1177 list_x_subtype_checker.ensure_type(v, path)
1178 else:
1179 x_subtype_checker.ensure_type(v, path)
1181 return AttributeTypeHandler(
1182 type_description,
1183 _validator,
1184 mapper=type_normalizer,
1185 )
1186 else:
1187 subtype_checker = [
1188 self._as_type_validator(attribute, a, parsing_typed_dict_attribute)
1189 for a in t_unmapped_args
1190 ]
1191 type_description = "one-of: " + ", ".join(
1192 f"{sc.describe_type()}" for sc in subtype_checker
1193 )
1194 mapper = subtype_checker[0].mapper
1195 if any(mapper != sc.mapper for sc in subtype_checker): 1195 ↛ 1196line 1195 didn't jump to line 1196 because the condition on line 1195 was never true
1196 raise ValueError(
1197 f'Cannot handle the union "{provided_type}" as the target types need different'
1198 " type normalization/mapping logic. Unions are generally limited to Union[X, List[X]]"
1199 " where X is a non-collection type."
1200 )
1202 def _validator(v: Any, path: AttributePath) -> None:
1203 partial_matches = []
1204 for sc in subtype_checker: 1204 ↛ 1212line 1204 didn't jump to line 1212 because the loop on line 1204 didn't complete
1205 try:
1206 sc.ensure_type(v, path)
1207 return
1208 except ManifestParseException as e:
1209 if sc.base_type_match(v): 1209 ↛ 1210line 1209 didn't jump to line 1210 because the condition on line 1209 was never true
1210 partial_matches.append((sc, e))
1212 if len(partial_matches) == 1:
1213 raise partial_matches[0][1]
1214 _validation_type_error(
1215 path, f"Could not match against: {type_description}"
1216 )
1218 return AttributeTypeHandler(
1219 type_description,
1220 _validator,
1221 mapper=type_normalizer,
1222 )
1223 if t_unmapped_orig == Literal:
1224 # We want "x" for string values; repr provides 'x'
1225 pretty = ", ".join(
1226 f"`{v}`" if isinstance(v, str) else str(v) for v in t_unmapped_args
1227 )
1229 def _validator(v: Any, path: AttributePath) -> None:
1230 if v not in t_unmapped_args:
1231 value_hint = ""
1232 if isinstance(v, str):
1233 value_hint = f"({v}) "
1234 _validation_type_error(
1235 path,
1236 f"Value {value_hint}must be one of the following literal values: {pretty}",
1237 )
1239 return AttributeTypeHandler(
1240 f"One of the following literal values: {pretty}",
1241 _validator,
1242 )
1244 if provided_type == Any: 1244 ↛ 1249line 1244 didn't jump to line 1249 because the condition on line 1244 was always true
1245 return AttributeTypeHandler(
1246 "any (unvalidated)",
1247 lambda *a: None,
1248 )
1249 raise ValueError(
1250 f'The attribute "{attribute}" had/contained a type {provided_type}, which is not supported'
1251 )
1253 def _parse_types(
1254 self,
1255 spec: type[TypedDict],
1256 allow_target_attribute_annotation: bool = False,
1257 allow_source_attribute_annotations: bool = False,
1258 forbid_optional: bool = True,
1259 ) -> dict[str, AttributeDescription]:
1260 annotations = get_type_hints(spec, include_extras=True)
1261 return {
1262 k: self._attribute_description(
1263 k,
1264 t,
1265 k in spec.__required_keys__,
1266 allow_target_attribute_annotation=allow_target_attribute_annotation,
1267 allow_source_attribute_annotations=allow_source_attribute_annotations,
1268 forbid_optional=forbid_optional,
1269 )
1270 for k, t in annotations.items()
1271 }
1273 def _attribute_description(
1274 self,
1275 attribute: str,
1276 orig_td: Any,
1277 is_required: bool,
1278 forbid_optional: bool = True,
1279 allow_target_attribute_annotation: bool = False,
1280 allow_source_attribute_annotations: bool = False,
1281 ) -> AttributeDescription:
1282 td, anno, is_optional = _parse_type(
1283 attribute, orig_td, forbid_optional=forbid_optional
1284 )
1285 type_validator = self._as_type_validator(attribute, td, True)
1286 parsed_annotations = DetectedDebputyParseHint.parse_annotations(
1287 anno,
1288 f' Seen with attribute "{attribute}".',
1289 attribute,
1290 is_required,
1291 allow_target_attribute_annotation=allow_target_attribute_annotation,
1292 allow_source_attribute_annotations=allow_source_attribute_annotations,
1293 )
1294 return AttributeDescription(
1295 target_attribute=parsed_annotations.target_attribute,
1296 attribute_type=td,
1297 type_validator=type_validator,
1298 annotations=anno,
1299 is_optional=is_optional,
1300 conflicting_attributes=parsed_annotations.conflict_with_source_attributes,
1301 conditional_required=parsed_annotations.conditional_required,
1302 source_attribute_name=assume_not_none(
1303 parsed_annotations.source_manifest_attribute
1304 ),
1305 parse_hints=parsed_annotations,
1306 )
1308 def _parse_alt_form(
1309 self,
1310 alt_form,
1311 default_target_attribute: str | None,
1312 ) -> AttributeDescription:
1313 td, anno, is_optional = _parse_type(
1314 "source_format alternative form",
1315 alt_form,
1316 forbid_optional=True,
1317 parsing_typed_dict_attribute=False,
1318 )
1319 type_validator = self._as_type_validator(
1320 "source_format alternative form",
1321 td,
1322 True,
1323 )
1324 parsed_annotations = DetectedDebputyParseHint.parse_annotations(
1325 anno,
1326 " The alternative for source_format.",
1327 None,
1328 False,
1329 default_target_attribute=default_target_attribute,
1330 allow_target_attribute_annotation=True,
1331 allow_source_attribute_annotations=False,
1332 )
1333 return AttributeDescription(
1334 target_attribute=parsed_annotations.target_attribute,
1335 attribute_type=td,
1336 type_validator=type_validator,
1337 annotations=anno,
1338 is_optional=is_optional,
1339 conflicting_attributes=parsed_annotations.conflict_with_source_attributes,
1340 conditional_required=parsed_annotations.conditional_required,
1341 source_attribute_name="Alt form of the source_format",
1342 )
1344 def _union_narrowing(
1345 self,
1346 input_type: Any,
1347 target_type: Any,
1348 parsing_typed_dict_attribute: bool,
1349 ) -> Callable[[Any, AttributePath, Optional["ParserContextData"]], Any] | None:
1350 _, input_orig, input_args = unpack_type(
1351 input_type, parsing_typed_dict_attribute
1352 )
1353 _, target_orig, target_args = unpack_type(
1354 target_type, parsing_typed_dict_attribute
1355 )
1357 if input_orig not in (Union, UnionType) or not input_args: 1357 ↛ 1358line 1357 didn't jump to line 1358 because the condition on line 1357 was never true
1358 raise ValueError("input_type must be a Union[...] with non-empty args")
1360 # Currently, we only support Union[X, List[X]] -> List[Y] narrowing or Union[X, List[X]] -> Union[Y, Union[Y]]
1361 # - Where X = Y or there is a simple standard transformation from X to Y.
1363 if target_orig not in (Union, UnionType, list) or not target_args:
1364 # Not supported
1365 return None
1367 if target_orig in (Union, UnionType) and set(input_args) == set(target_args): 1367 ↛ 1369line 1367 didn't jump to line 1369 because the condition on line 1367 was never true
1368 # Not needed (identity mapping)
1369 return None
1371 if target_orig == list and not any(get_origin(a) == list for a in input_args): 1371 ↛ 1373line 1371 didn't jump to line 1373 because the condition on line 1371 was never true
1372 # Not supported
1373 return None
1375 target_arg = target_args[0]
1376 simplified_type = self._strip_mapped_types(
1377 target_arg, parsing_typed_dict_attribute
1378 )
1379 acceptable_types = {
1380 target_arg,
1381 list[target_arg], # type: ignore
1382 List[target_arg], # type: ignore
1383 simplified_type,
1384 list[simplified_type], # type: ignore
1385 List[simplified_type], # type: ignore
1386 }
1387 target_format = (
1388 target_arg,
1389 list[target_arg], # type: ignore
1390 List[target_arg], # type: ignore
1391 )
1392 in_target_format = 0
1393 in_simple_format = 0
1394 for input_arg in input_args:
1395 if input_arg not in acceptable_types: 1395 ↛ 1397line 1395 didn't jump to line 1397 because the condition on line 1395 was never true
1396 # Not supported
1397 return None
1398 if input_arg in target_format:
1399 in_target_format += 1
1400 else:
1401 in_simple_format += 1
1403 assert in_simple_format or in_target_format
1405 if in_target_format and not in_simple_format:
1406 # Union[X, List[X]] -> List[X]
1407 return normalize_into_list
1408 mapped = self._registered_types[target_arg]
1409 if not in_target_format and in_simple_format: 1409 ↛ 1424line 1409 didn't jump to line 1424 because the condition on line 1409 was always true
1410 # Union[X, List[X]] -> List[Y]
1412 def _mapper_x_list_y(
1413 x: Any | list[Any],
1414 ap: AttributePath,
1415 pc: Optional["ParserContextData"],
1416 ) -> list[Any]:
1417 in_list_form: list[Any] = normalize_into_list(x, ap, pc)
1419 return [mapped.mapper(x, ap, pc) for x in in_list_form]
1421 return _mapper_x_list_y
1423 # Union[Y, List[X]] -> List[Y]
1424 if not isinstance(target_arg, type):
1425 raise ValueError(
1426 f"Cannot narrow {input_type} -> {target_type}: The automatic conversion does"
1427 f" not support mixed types. Please use either {simplified_type} or {target_arg}"
1428 f" in the source content (but both a mix of both)"
1429 )
1431 def _mapper_mixed_list_y(
1432 x: Any | list[Any],
1433 ap: AttributePath,
1434 pc: Optional["ParserContextData"],
1435 ) -> list[Any]:
1436 in_list_form: list[Any] = normalize_into_list(x, ap, pc)
1438 return [
1439 x if isinstance(x, target_arg) else mapped.mapper(x, ap, pc)
1440 for x in in_list_form
1441 ]
1443 return _mapper_mixed_list_y
1445 def _type_normalize(
1446 self,
1447 attribute: str,
1448 input_type: Any,
1449 target_type: Any,
1450 parsing_typed_dict_attribute: bool,
1451 ) -> Callable[[Any, AttributePath, Optional["ParserContextData"]], Any] | None:
1452 if input_type == target_type:
1453 return None
1454 _, input_orig, input_args = unpack_type(
1455 input_type, parsing_typed_dict_attribute
1456 )
1457 _, target_orig, target_args = unpack_type(
1458 target_type,
1459 parsing_typed_dict_attribute,
1460 )
1461 if input_orig in (Union, UnionType):
1462 result = self._union_narrowing(
1463 input_type, target_type, parsing_typed_dict_attribute
1464 )
1465 if result:
1466 return result
1467 elif target_orig == list and target_args[0] == input_type:
1468 return wrap_into_list
1470 mapped = self._registered_types.get(target_type)
1471 if mapped is not None and input_type == mapped.source_type:
1472 # Source -> Target
1473 return mapped.mapper
1474 if target_orig == list and target_args: 1474 ↛ 1492line 1474 didn't jump to line 1492 because the condition on line 1474 was always true
1475 mapped = self._registered_types.get(target_args[0])
1476 if mapped is not None: 1476 ↛ 1492line 1476 didn't jump to line 1492 because the condition on line 1476 was always true
1477 # mypy is dense and forgot `mapped` cannot be optional in the comprehensions.
1478 mapped_type: TypeMapping = mapped
1479 if input_type == mapped.source_type: 1479 ↛ 1481line 1479 didn't jump to line 1481 because the condition on line 1479 was never true
1480 # Source -> List[Target]
1481 return lambda x, ap, pc: [mapped_type.mapper(x, ap, pc)]
1482 if ( 1482 ↛ 1492line 1482 didn't jump to line 1492 because the condition on line 1482 was always true
1483 input_orig == list
1484 and input_args
1485 and input_args[0] == mapped_type.source_type
1486 ):
1487 # List[Source] -> List[Target]
1488 return lambda xs, ap, pc: [
1489 mapped_type.mapper(x, ap, pc) for x in xs
1490 ]
1492 raise ValueError(
1493 f'Unsupported type normalization for "{attribute}": Cannot automatically map/narrow'
1494 f" {input_type} to {target_type}"
1495 )
1497 def _strip_mapped_types(
1498 self, orig_td: Any, parsing_typed_dict_attribute: bool
1499 ) -> Any:
1500 m = self._registered_types.get(orig_td)
1501 if m is not None:
1502 return m.source_type
1503 _, v, args = unpack_type(orig_td, parsing_typed_dict_attribute)
1504 if v == list:
1505 arg = args[0]
1506 m = self._registered_types.get(arg)
1507 if m:
1508 return list[m.source_type] # type: ignore
1509 if v in (Union, UnionType):
1510 stripped_args = tuple(
1511 self._strip_mapped_types(x, parsing_typed_dict_attribute) for x in args
1512 )
1513 if stripped_args != args:
1514 return Union[stripped_args]
1515 return orig_td
1518def _sort_key(attr: StandardParserAttributeDocumentation) -> Any:
1519 key = next(iter(attr.attributes))
1520 return attr.sort_category, key
1523def _apply_std_docs(
1524 std_doc_table: (
1525 Mapping[type[Any], Sequence[StandardParserAttributeDocumentation]] | None
1526 ),
1527 source_format_typed_dict: type[Any],
1528 attribute_docs: Sequence[ParserAttributeDocumentation] | None,
1529) -> Sequence[ParserAttributeDocumentation] | None:
1530 if std_doc_table is None or not std_doc_table: 1530 ↛ 1533line 1530 didn't jump to line 1533 because the condition on line 1530 was always true
1531 return attribute_docs
1533 has_docs_for = set()
1534 if attribute_docs:
1535 for attribute_doc in attribute_docs:
1536 has_docs_for.update(attribute_doc.attributes)
1538 base_seen = set()
1539 std_docs_used = []
1541 remaining_bases = set(getattr(source_format_typed_dict, "__orig_bases__", []))
1542 base_seen.update(remaining_bases)
1543 while remaining_bases:
1544 base = remaining_bases.pop()
1545 new_bases_to_check = {
1546 x for x in getattr(base, "__orig_bases__", []) if x not in base_seen
1547 }
1548 remaining_bases.update(new_bases_to_check)
1549 base_seen.update(new_bases_to_check)
1550 std_docs = std_doc_table.get(base)
1551 if std_docs:
1552 for std_doc in std_docs:
1553 if any(a in has_docs_for for a in std_doc.attributes):
1554 # If there is any overlap, do not add the docs
1555 continue
1556 has_docs_for.update(std_doc.attributes)
1557 std_docs_used.append(std_doc)
1559 if not std_docs_used:
1560 return attribute_docs
1561 docs = sorted(std_docs_used, key=_sort_key)
1562 if attribute_docs:
1563 # Plugin provided attributes first
1564 c = list(attribute_docs)
1565 c.extend(docs)
1566 docs = c
1567 return tuple(docs)
1570def _verify_and_auto_correct_inline_reference_documentation(
1571 parsed_content: type[TD],
1572 source_typed_dict: type[Any],
1573 source_content_attributes: Mapping[str, AttributeDescription],
1574 inline_reference_documentation: ParserDocumentation | None,
1575 has_alt_form: bool,
1576 automatic_docs: (
1577 Mapping[type[Any], Sequence[StandardParserAttributeDocumentation]] | None
1578 ) = None,
1579) -> ParserDocumentation | None:
1580 orig_attribute_docs = (
1581 inline_reference_documentation.attribute_doc
1582 if inline_reference_documentation
1583 else None
1584 )
1585 attribute_docs = _apply_std_docs(
1586 automatic_docs,
1587 source_typed_dict,
1588 orig_attribute_docs,
1589 )
1590 if inline_reference_documentation is None and attribute_docs is None:
1591 return None
1592 changes = {}
1593 if attribute_docs:
1594 seen = set()
1595 had_any_custom_docs = False
1596 for attr_doc in attribute_docs:
1597 if not isinstance(attr_doc, StandardParserAttributeDocumentation):
1598 had_any_custom_docs = True
1599 for attr_name in attr_doc.attributes:
1600 attr = source_content_attributes.get(attr_name)
1601 if attr is None: 1601 ↛ 1602line 1601 didn't jump to line 1602 because the condition on line 1601 was never true
1602 raise ValueError(
1603 f"The inline_reference_documentation for the source format of {parsed_content.__qualname__}"
1604 f' references an attribute "{attr_name}", which does not exist in the source format.'
1605 )
1606 if attr_name in seen: 1606 ↛ 1607line 1606 didn't jump to line 1607 because the condition on line 1606 was never true
1607 raise ValueError(
1608 f"The inline_reference_documentation for the source format of {parsed_content.__qualname__}"
1609 f' has documentation for "{attr_name}" twice, which is not supported.'
1610 f" Please document it at most once"
1611 )
1612 seen.add(attr_name)
1613 undocumented = source_content_attributes.keys() - seen
1614 if undocumented: 1614 ↛ 1615line 1614 didn't jump to line 1615 because the condition on line 1614 was never true
1615 if had_any_custom_docs:
1616 undocumented_attrs = ", ".join(undocumented)
1617 raise ValueError(
1618 f"The following attributes were not documented for the source format of"
1619 f" {parsed_content.__qualname__}. If this is deliberate, then please"
1620 ' declare each them as undocumented (via undocumented_attr("foo")):'
1621 f" {undocumented_attrs}"
1622 )
1623 combined_docs = list(attribute_docs)
1624 combined_docs.extend(undocumented_attr(a) for a in sorted(undocumented))
1625 attribute_docs = combined_docs
1627 if attribute_docs and orig_attribute_docs != attribute_docs: 1627 ↛ 1628line 1627 didn't jump to line 1628 because the condition on line 1627 was never true
1628 assert attribute_docs is not None
1629 changes["attribute_doc"] = tuple(attribute_docs)
1631 if ( 1631 ↛ 1636line 1631 didn't jump to line 1636 because the condition on line 1631 was never true
1632 inline_reference_documentation is not None
1633 and inline_reference_documentation.alt_parser_description
1634 and not has_alt_form
1635 ):
1636 raise ValueError(
1637 "The inline_reference_documentation had documentation for an non-mapping format,"
1638 " but the source format does not have a non-mapping format."
1639 )
1640 if changes: 1640 ↛ 1641line 1640 didn't jump to line 1641 because the condition on line 1640 was never true
1641 if inline_reference_documentation is None:
1642 inline_reference_documentation = reference_documentation()
1643 return inline_reference_documentation.replace(**changes)
1644 return inline_reference_documentation
1647def _check_conflicts(
1648 input_content_attributes: dict[str, AttributeDescription],
1649 required_attributes: frozenset[str],
1650 all_attributes: frozenset[str],
1651) -> None:
1652 for attr_name, attr in input_content_attributes.items():
1653 if attr_name in required_attributes and attr.conflicting_attributes: 1653 ↛ 1654line 1653 didn't jump to line 1654 because the condition on line 1653 was never true
1654 c = ", ".join(repr(a) for a in attr.conflicting_attributes)
1655 raise ValueError(
1656 f'The attribute "{attr_name}" is required and conflicts with the attributes: {c}.'
1657 " This makes it impossible to use these attributes. Either remove the attributes"
1658 f' (along with the conflicts for them), adjust the conflicts or make "{attr_name}"'
1659 " optional (NotRequired)"
1660 )
1661 else:
1662 required_conflicts = attr.conflicting_attributes & required_attributes
1663 if required_conflicts: 1663 ↛ 1664line 1663 didn't jump to line 1664 because the condition on line 1663 was never true
1664 c = ", ".join(repr(a) for a in required_conflicts)
1665 raise ValueError(
1666 f'The attribute "{attr_name}" conflicts with the following *required* attributes: {c}.'
1667 f' This makes it impossible to use the "{attr_name}" attribute. Either remove it,'
1668 f" adjust the conflicts or make the listed attributes optional (NotRequired)"
1669 )
1670 unknown_attributes = attr.conflicting_attributes - all_attributes
1671 if unknown_attributes: 1671 ↛ 1672line 1671 didn't jump to line 1672 because the condition on line 1671 was never true
1672 c = ", ".join(repr(a) for a in unknown_attributes)
1673 raise ValueError(
1674 f'The attribute "{attr_name}" declares a conflict with the following unknown attributes: {c}.'
1675 f" None of these attributes were declared in the input."
1676 )
1679def _check_attributes(
1680 content: type[TypedDict],
1681 input_content: type[TypedDict],
1682 input_content_attributes: dict[str, AttributeDescription],
1683 sources: Mapping[str, Collection[str]],
1684) -> None:
1685 target_required_keys = content.__required_keys__
1686 input_required_keys = input_content.__required_keys__
1687 all_input_keys = input_required_keys | input_content.__optional_keys__
1689 for input_name in all_input_keys:
1690 attr = input_content_attributes[input_name]
1691 target_name = attr.target_attribute
1692 source_names = sources[target_name]
1693 input_is_required = input_name in input_required_keys
1694 target_is_required = target_name in target_required_keys
1696 assert source_names
1698 if input_is_required and len(source_names) > 1: 1698 ↛ 1699line 1698 didn't jump to line 1699 because the condition on line 1698 was never true
1699 raise ValueError(
1700 f'The source attribute "{input_name}" is required, but it maps to "{target_name}",'
1701 f' which has multiple sources "{source_names}". If "{input_name}" should be required,'
1702 f' then there is no need for additional sources for "{target_name}". Alternatively,'
1703 f' "{input_name}" might be missing a NotRequired type'
1704 f' (example: "{input_name}: NotRequired[<OriginalTypeHere>]")'
1705 )
1706 if not input_is_required and target_is_required and len(source_names) == 1: 1706 ↛ 1707line 1706 didn't jump to line 1707 because the condition on line 1706 was never true
1707 raise ValueError(
1708 f'The source attribute "{input_name}" is not marked as required and maps to'
1709 f' "{target_name}", which is marked as required. As there are no other attributes'
1710 f' mapping to "{target_name}", then "{input_name}" must be required as well'
1711 f' ("{input_name}: Required[<Type>]"). Alternatively, "{target_name}" should be optional'
1712 f' ("{target_name}: NotRequired[<Type>]") or an "MappingHint.aliasOf" might be missing.'
1713 )
1716def _validation_type_error(path: AttributePath, message: str) -> None:
1717 raise ManifestParseException(
1718 f'The attribute "{path.path}" did not have a valid structure/type: {message}'
1719 )
1722def _is_two_arg_x_list_x(t_args: tuple[Any, ...]) -> bool:
1723 if len(t_args) != 2:
1724 return False
1725 lhs, rhs = t_args
1726 if get_origin(lhs) == list:
1727 if get_origin(rhs) == list: 1727 ↛ 1730line 1727 didn't jump to line 1730 because the condition on line 1727 was never true
1728 # It could still match X, List[X] - but we do not allow this case for now as the caller
1729 # does not support it.
1730 return False
1731 l_args = get_args(lhs)
1732 return bool(l_args and l_args[0] == rhs)
1733 if get_origin(rhs) == list:
1734 r_args = get_args(rhs)
1735 return bool(r_args and r_args[0] == lhs)
1736 return False
1739def _extract_typed_dict(
1740 base_type,
1741 default_target_attribute: str | None,
1742) -> tuple[type[TypedDict] | None, Any]:
1743 if is_typeddict(base_type):
1744 return base_type, None
1745 _, origin, args = unpack_type(base_type, False)
1746 if origin != Union:
1747 if isinstance(base_type, type) and issubclass(base_type, (dict, Mapping)): 1747 ↛ 1748line 1747 didn't jump to line 1748 because the condition on line 1747 was never true
1748 raise ValueError(
1749 "The source_format cannot be nor contain a (non-TypedDict) dict"
1750 )
1751 return None, base_type
1752 typed_dicts = [x for x in args if is_typeddict(x)]
1753 if len(typed_dicts) > 1: 1753 ↛ 1754line 1753 didn't jump to line 1754 because the condition on line 1753 was never true
1754 raise ValueError(
1755 "When source_format is a Union, it must contain at most one TypedDict"
1756 )
1757 typed_dict = typed_dicts[0] if typed_dicts else None
1759 if any(x is None or x is _NONE_TYPE for x in args): 1759 ↛ 1760line 1759 didn't jump to line 1760 because the condition on line 1759 was never true
1760 raise ValueError(
1761 "The source_format cannot be nor contain Optional[X] or Union[X, None]"
1762 )
1764 if any( 1764 ↛ 1769line 1764 didn't jump to line 1769 because the condition on line 1764 was never true
1765 isinstance(x, type) and issubclass(x, (dict, Mapping))
1766 for x in args
1767 if x is not typed_dict
1768 ):
1769 raise ValueError(
1770 "The source_format cannot be nor contain a (non-TypedDict) dict"
1771 )
1772 remaining = [x for x in args if x is not typed_dict]
1773 has_target_attribute = False
1774 anno = None
1775 if len(remaining) == 1: 1775 ↛ 1776line 1775 didn't jump to line 1776 because the condition on line 1775 was never true
1776 base_type, anno, _ = _parse_type(
1777 "source_format alternative form",
1778 remaining[0],
1779 forbid_optional=True,
1780 parsing_typed_dict_attribute=False,
1781 )
1782 has_target_attribute = bool(anno) and any(
1783 isinstance(x, TargetAttribute) for x in anno
1784 )
1785 target_type = base_type
1786 else:
1787 target_type = Union[tuple(remaining)]
1789 if default_target_attribute is None and not has_target_attribute: 1789 ↛ 1790line 1789 didn't jump to line 1790 because the condition on line 1789 was never true
1790 raise ValueError(
1791 'The alternative format must be Union[TypedDict,Annotated[X, DebputyParseHint.target_attribute("...")]]'
1792 " OR the parsed_content format must have exactly one attribute that is required."
1793 )
1794 if anno: 1794 ↛ 1795line 1794 didn't jump to line 1795 because the condition on line 1794 was never true
1795 final_anno = [target_type]
1796 final_anno.extend(anno)
1797 return typed_dict, Annotated[tuple(final_anno)]
1798 return typed_dict, target_type
1801def _dispatch_parse_generator(
1802 dispatch_type: type[DebputyDispatchableType],
1803) -> Callable[[Any, AttributePath, Optional["ParserContextData"]], Any]:
1804 def _dispatch_parse(
1805 value: Any,
1806 attribute_path: AttributePath,
1807 parser_context: Optional["ParserContextData"],
1808 ):
1809 assert parser_context is not None
1810 dispatching_parser = parser_context.dispatch_parser_table_for(dispatch_type)
1811 return dispatching_parser.parse_input(
1812 value, attribute_path, parser_context=parser_context
1813 )
1815 return _dispatch_parse
1818def _dispatch_parser(
1819 dispatch_type: type[DebputyDispatchableType],
1820) -> AttributeTypeHandler:
1821 return AttributeTypeHandler(
1822 dispatch_type.__name__,
1823 lambda *a: None,
1824 mapper=_dispatch_parse_generator(dispatch_type),
1825 )
1828def _parse_type(
1829 attribute: str,
1830 orig_td: Any,
1831 forbid_optional: bool = True,
1832 parsing_typed_dict_attribute: bool = True,
1833) -> tuple[Any, tuple[Any, ...], bool]:
1834 td, v, args = unpack_type(orig_td, parsing_typed_dict_attribute)
1835 md: tuple[Any, ...] = ()
1836 optional = False
1837 if v is not None:
1838 if v == Annotated:
1839 anno = get_args(td)
1840 md = anno[1:]
1841 td, v, args = unpack_type(anno[0], parsing_typed_dict_attribute)
1843 if td is _NONE_TYPE: 1843 ↛ 1844line 1843 didn't jump to line 1844 because the condition on line 1843 was never true
1844 raise ValueError(
1845 f'The attribute "{attribute}" resolved to type "None". "Nil" / "None" fields are not allowed in the'
1846 " debputy manifest, so this attribute does not make sense in its current form."
1847 )
1848 if ( 1848 ↛ 1853line 1848 didn't jump to line 1853 because the condition on line 1848 was never true
1849 forbid_optional
1850 and (v == Union or v == types.UnionType)
1851 and any(a is _NONE_TYPE for a in args)
1852 ):
1853 raise ValueError(
1854 f'Detected use of Optional in "{attribute}", which is not allowed here.'
1855 " Please use NotRequired for optional fields"
1856 )
1858 return td, md, optional
1861def _normalize_attribute_name(attribute: str) -> str:
1862 if attribute.endswith("_"):
1863 attribute = attribute[:-1]
1864 return attribute.replace("_", "-")
1867@dataclasses.dataclass
1868class DetectedDebputyParseHint:
1869 target_attribute: str
1870 source_manifest_attribute: str | None
1871 conflict_with_source_attributes: frozenset[str]
1872 conditional_required: ConditionalRequired | None
1873 applicable_as_path_hint: bool
1875 @classmethod
1876 def parse_annotations(
1877 cls,
1878 anno: tuple[Any, ...],
1879 error_context: str,
1880 default_attribute_name: str | None,
1881 is_required: bool,
1882 default_target_attribute: str | None = None,
1883 allow_target_attribute_annotation: bool = False,
1884 allow_source_attribute_annotations: bool = False,
1885 ) -> "DetectedDebputyParseHint":
1886 target_attr_anno = find_annotation(anno, TargetAttribute)
1887 if target_attr_anno:
1888 if not allow_target_attribute_annotation: 1888 ↛ 1889line 1888 didn't jump to line 1889 because the condition on line 1888 was never true
1889 raise ValueError(
1890 f"The DebputyParseHint.target_attribute annotation is not allowed in this context.{error_context}"
1891 )
1892 target_attribute = target_attr_anno.attribute
1893 elif default_target_attribute is not None:
1894 target_attribute = default_target_attribute
1895 elif default_attribute_name is not None: 1895 ↛ 1898line 1895 didn't jump to line 1898 because the condition on line 1895 was always true
1896 target_attribute = default_attribute_name
1897 else:
1898 if default_attribute_name is None:
1899 raise ValueError(
1900 "allow_target_attribute_annotation must be True OR "
1901 "default_attribute_name/default_target_attribute must be not None"
1902 )
1903 raise ValueError(
1904 f"Missing DebputyParseHint.target_attribute annotation.{error_context}"
1905 )
1906 source_attribute_anno = find_annotation(anno, ManifestAttribute)
1907 _source_attribute_allowed(
1908 allow_source_attribute_annotations, error_context, source_attribute_anno
1909 )
1910 if source_attribute_anno:
1911 source_attribute_name = source_attribute_anno.attribute
1912 elif default_attribute_name is not None:
1913 source_attribute_name = _normalize_attribute_name(default_attribute_name)
1914 else:
1915 source_attribute_name = None
1916 mutual_exclusive_with_anno = find_annotation(anno, ConflictWithSourceAttribute)
1917 if mutual_exclusive_with_anno:
1918 _source_attribute_allowed(
1919 allow_source_attribute_annotations,
1920 error_context,
1921 mutual_exclusive_with_anno,
1922 )
1923 conflicting_attributes = mutual_exclusive_with_anno.conflicting_attributes
1924 else:
1925 conflicting_attributes = frozenset()
1926 conditional_required = find_annotation(anno, ConditionalRequired)
1928 if conditional_required and is_required: 1928 ↛ 1929line 1928 didn't jump to line 1929 because the condition on line 1928 was never true
1929 if default_attribute_name is None:
1930 raise ValueError(
1931 "is_required cannot be True without default_attribute_name being not None"
1932 )
1933 raise ValueError(
1934 f'The attribute "{default_attribute_name}" is Required while also being conditionally required.'
1935 ' Please make the attribute "NotRequired" or remove the conditional requirement.'
1936 )
1938 not_path_hint_anno = find_annotation(anno, NotPathHint)
1939 applicable_as_path_hint = not_path_hint_anno is None
1941 return DetectedDebputyParseHint(
1942 target_attribute=target_attribute,
1943 source_manifest_attribute=source_attribute_name,
1944 conflict_with_source_attributes=conflicting_attributes,
1945 conditional_required=conditional_required,
1946 applicable_as_path_hint=applicable_as_path_hint,
1947 )
1950def _source_attribute_allowed(
1951 source_attribute_allowed: bool,
1952 error_context: str,
1953 annotation: DebputyParseHint | None,
1954) -> None:
1955 if source_attribute_allowed or annotation is None: 1955 ↛ 1957line 1955 didn't jump to line 1957 because the condition on line 1955 was always true
1956 return
1957 raise ValueError(
1958 f'The annotation "{annotation}" cannot be used here. {error_context}'
1959 )
1962def _handle_str_enum(enum_class: type[enum.StrEnum]) -> AttributeTypeHandler:
1963 pretty = ", ".join(f"`{e}`" for e in enum_class)
1965 def _validator(v: Any, path: AttributePath) -> None:
1966 valid = False
1967 value_hint = ""
1968 if isinstance(v, str):
1969 value_hint = f"({v}) "
1970 try:
1971 enum_class(v)
1972 valid = True
1973 except ValueError:
1974 pass
1975 if not valid:
1976 _validation_type_error(
1977 path,
1978 f"Value {value_hint}must be one of the following literal values: {pretty}",
1979 )
1981 def _mapper(
1982 v: Any, _path: AttributePath, _pc: ParserContextData | None
1983 ) -> enum.StrEnum:
1984 return enum_class(v)
1986 return AttributeTypeHandler(
1987 f"One of the following literal values: {pretty}",
1988 _validator,
1989 base_type=str,
1990 mapper=_mapper,
1991 )