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