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