Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/cattrs/gen/__init__.py: 6%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import re
4from collections.abc import Callable, Iterable, Mapping
5from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar
7from attrs import NOTHING, Attribute, Converter, Factory, evolve
8from typing_extensions import NoDefault
10from .._compat import (
11 ANIES,
12 TypeAlias,
13 adapted_fields,
14 get_args,
15 get_origin,
16 is_annotated,
17 is_bare,
18 is_bare_final,
19 is_generic,
20)
21from .._generics import deep_copy_with
22from ..dispatch import UnstructureHook
23from ..errors import (
24 AttributeValidationNote,
25 ClassValidationError,
26 ForbiddenExtraKeysError,
27 IterableValidationError,
28 IterableValidationNote,
29 StructureHandlerNotFoundError,
30)
31from ..fns import identity
32from ..types import SimpleStructureHook
33from ._consts import AttributeOverride, already_generating, neutral
34from ._generics import generate_mapping
35from ._lc import generate_unique_filename
36from ._shared import _annotated_override_or_default, find_structure_handler
38if TYPE_CHECKING:
39 from ..converters import BaseConverter
41__all__ = [
42 "make_dict_structure_fn",
43 "make_dict_structure_fn_from_attrs",
44 "make_dict_unstructure_fn",
45 "make_dict_unstructure_fn_from_attrs",
46 "make_hetero_tuple_structure_fn",
47 "make_hetero_tuple_unstructure_fn",
48 "make_iterable_unstructure_fn",
49 "make_mapping_structure_fn",
50 "make_mapping_unstructure_fn",
51]
54def override(
55 omit_if_default: bool | None = None,
56 rename: str | None = None,
57 omit: bool | None = None,
58 struct_hook: Callable[[Any, Any], Any] | None = None,
59 unstruct_hook: Callable[[Any], Any] | None = None,
60) -> AttributeOverride:
61 """Override how a particular field is handled.
63 :param omit: Whether to skip the field or not. `None` means apply default handling.
64 """
65 return AttributeOverride(omit_if_default, rename, omit, struct_hook, unstruct_hook)
68T = TypeVar("T")
71def make_dict_unstructure_fn_from_attrs(
72 attrs: list[Attribute],
73 cl: type[T],
74 converter: BaseConverter,
75 typevar_map: dict[str, Any] = {},
76 _cattrs_omit_if_default: bool = False,
77 _cattrs_use_linecache: bool = True,
78 _cattrs_use_alias: bool | Literal["from_converter"] = "from_converter",
79 _cattrs_include_init_false: bool = False,
80 **kwargs: AttributeOverride,
81) -> Callable[[T], dict[str, Any]]:
82 """
83 Generate a specialized dict unstructuring function for a list of attributes.
85 Usually used as a building block by more specialized hook factories.
87 Any provided overrides are attached to the generated function under the
88 `overrides` attribute.
90 :param cl: The class for which the function is generated; used mostly for its name,
91 module name and qualname.
92 :param _cattrs_omit_if_default: if true, attributes equal to their default values
93 will be omitted in the result dictionary.
94 :param _cattrs_use_alias: If true, the attribute alias will be used as the
95 dictionary key by default.
96 :param _cattrs_include_init_false: If true, _attrs_ fields marked as `init=False`
97 will be included.
99 .. versionadded:: 24.1.0
100 .. versionchanged:: 25.2.0
101 The `_cattrs_use_alias` parameter takes its value from the given converter
102 by default.
103 .. versionchanged:: 26.1.0
104 `typing.Annotated[T, override()]` is now recognized and can be used to customize
105 unstructuring.
106 .. versionchanged:: 26.1.0
107 When `_cattrs_omit_if_default` is true and the attribute has an attrs converter
108 specified, the converter is applied to the default value before checking if it
109 is equal to the attribute's value.
110 """
112 fn_name = "unstructure_" + cl.__name__
113 globs = {}
114 lines = []
115 invocation_lines = []
116 internal_arg_parts = {}
118 if _cattrs_use_alias == "from_converter":
119 # BaseConverter doesn't have it so we're careful.
120 _cattrs_use_alias = getattr(converter, "use_alias", False)
122 for a in attrs:
123 attr_name = a.name
124 if attr_name in kwargs:
125 override = kwargs[attr_name]
126 else:
127 override = _annotated_override_or_default(a.type, neutral)
128 if override != neutral:
129 kwargs[attr_name] = override
131 if override.omit:
132 continue
133 if override.omit is None and not a.init and not _cattrs_include_init_false:
134 continue
135 if override.rename is None:
136 kn = attr_name if not _cattrs_use_alias else a.alias
137 if kn != attr_name:
138 kwargs[attr_name] = evolve(override, rename=kn)
139 else:
140 kn = override.rename
141 d = a.default
143 # For each attribute, we try resolving the type here and now.
144 # If a type is manually overwritten, this function should be
145 # regenerated.
146 handler = None
147 if override.unstruct_hook is not None:
148 handler = override.unstruct_hook
149 else:
150 if a.type is not None:
151 t = a.type
152 if isinstance(t, TypeVar):
153 if t.__name__ in typevar_map:
154 t = typevar_map[t.__name__]
155 else:
156 handler = converter.unstructure
157 elif is_generic(t) and not is_bare(t) and not is_annotated(t):
158 t = deep_copy_with(t, typevar_map, cl)
160 if handler is None:
161 if (
162 is_bare_final(t)
163 and a.default is not NOTHING
164 and not isinstance(a.default, Factory)
165 ):
166 # This is a special case where we can use the
167 # type of the default to dispatch on.
168 t = a.default.__class__
169 try:
170 handler = converter.get_unstructure_hook(t, cache_result=False)
171 except RecursionError:
172 # There's a circular reference somewhere down the line
173 handler = converter.unstructure
174 else:
175 handler = converter.unstructure
177 is_identity = handler == identity
179 if not is_identity:
180 unstruct_handler_name = f"__c_unstr_{attr_name}"
181 globs[unstruct_handler_name] = handler
182 internal_arg_parts[unstruct_handler_name] = handler
183 invoke = f"{unstruct_handler_name}(instance.{attr_name})"
184 else:
185 invoke = f"instance.{attr_name}"
187 if d is not NOTHING and (
188 (_cattrs_omit_if_default and override.omit_if_default is not False)
189 or override.omit_if_default
190 ):
191 def_name = f"__c_def_{attr_name}"
193 if isinstance(d, Factory):
194 globs[def_name] = d.factory
195 internal_arg_parts[def_name] = d.factory
196 def_str = f"{def_name}(instance)" if d.takes_self else f"{def_name}()"
197 else:
198 globs[def_name] = d
199 internal_arg_parts[def_name] = d
200 def_str = def_name
202 c = a.converter
203 if c is not None:
204 conv_name = f"__c_conv_{attr_name}"
205 if isinstance(c, Converter):
206 globs[conv_name] = c
207 internal_arg_parts[conv_name] = c
208 field_name = f"__c_field_{attr_name}"
209 globs[field_name] = a
210 internal_arg_parts[field_name] = a
211 def_str = f"{conv_name}({def_str}, instance, {field_name})"
212 elif isinstance(d, Factory):
213 globs[conv_name] = c
214 internal_arg_parts[conv_name] = c
215 def_str = f"{conv_name}({def_str})"
216 else:
217 globs[def_name] = c(d)
218 internal_arg_parts[def_name] = c(d)
220 lines.append(f" if instance.{attr_name} != {def_str}:")
221 lines.append(f" res[{kn!r}] = {invoke}")
223 else:
224 # No default or no override.
225 invocation_lines.append(f"{kn!r}: {invoke},")
227 internal_arg_line = ", ".join([f"{i}={i}" for i in internal_arg_parts])
228 if internal_arg_line:
229 internal_arg_line = f", {internal_arg_line}"
230 for k, v in internal_arg_parts.items():
231 globs[k] = v
233 total_lines = (
234 [f"def {fn_name}(instance{internal_arg_line}):"]
235 + [" res = {"]
236 + [f" {line}" for line in invocation_lines]
237 + [" }"]
238 + lines
239 + [" return res"]
240 )
241 script = "\n".join(total_lines)
242 fname = generate_unique_filename(
243 cl, "unstructure", lines=total_lines if _cattrs_use_linecache else []
244 )
246 eval(compile(script, fname, "exec"), globs)
248 res = globs[fn_name]
249 res.overrides = kwargs
251 return res
254def make_dict_unstructure_fn(
255 cl: type[T],
256 converter: BaseConverter,
257 _cattrs_omit_if_default: bool = False,
258 _cattrs_use_linecache: bool = True,
259 _cattrs_use_alias: bool | Literal["from_converter"] = "from_converter",
260 _cattrs_include_init_false: bool = False,
261 **kwargs: AttributeOverride,
262) -> Callable[[T], dict[str, Any]]:
263 """
264 Generate a specialized dict unstructuring function for an attrs class or a
265 dataclass.
267 Any provided overrides are attached to the generated function under the
268 `overrides` attribute.
270 :param _cattrs_omit_if_default: if true, attributes equal to their default values
271 will be omitted in the result dictionary.
272 :param _cattrs_use_alias: If true, the attribute alias will be used as the
273 dictionary key by default.
274 :param _cattrs_include_init_false: If true, _attrs_ fields marked as `init=False`
275 will be included.
277 .. versionadded:: 23.2.0 *_cattrs_use_alias*
278 .. versionadded:: 23.2.0 *_cattrs_include_init_false*
279 .. versionchanged:: 25.2.0
280 The `_cattrs_use_alias` parameter takes its value from the given converter
281 by default.
282 .. versionchanged:: 26.1.0
283 `typing.Annotated[T, override()]` is now recognized and can be used to customize
284 unstructuring.
285 """
286 origin = get_origin(cl)
287 attrs = adapted_fields(origin or cl) # type: ignore
289 mapping = {}
290 if _cattrs_use_alias == "from_converter":
291 # BaseConverter doesn't have it so we're careful.
292 _cattrs_use_alias = getattr(converter, "use_alias", False)
293 if is_generic(cl):
294 mapping = generate_mapping(cl, mapping)
296 if origin is not None:
297 cl = origin
299 # We keep track of what we're generating to help with recursive
300 # class graphs.
301 try:
302 working_set = already_generating.working_set
303 except AttributeError:
304 working_set = set()
305 already_generating.working_set = working_set
306 if cl in working_set:
307 raise RecursionError()
309 working_set.add(cl)
311 try:
312 return make_dict_unstructure_fn_from_attrs(
313 attrs,
314 cl,
315 converter,
316 mapping,
317 _cattrs_omit_if_default=_cattrs_omit_if_default,
318 _cattrs_use_linecache=_cattrs_use_linecache,
319 _cattrs_use_alias=_cattrs_use_alias,
320 _cattrs_include_init_false=_cattrs_include_init_false,
321 **kwargs,
322 )
323 finally:
324 working_set.remove(cl)
325 if not working_set:
326 del already_generating.working_set
329def make_dict_structure_fn_from_attrs(
330 attrs: list[Attribute],
331 cl: type[T],
332 converter: BaseConverter,
333 typevar_map: dict[str, Any] = {},
334 _cattrs_forbid_extra_keys: bool | Literal["from_converter"] = "from_converter",
335 _cattrs_use_linecache: bool = True,
336 _cattrs_prefer_attrib_converters: (
337 bool | Literal["from_converter"]
338 ) = "from_converter",
339 _cattrs_detailed_validation: bool | Literal["from_converter"] = "from_converter",
340 _cattrs_use_alias: bool | Literal["from_converter"] = "from_converter",
341 _cattrs_include_init_false: bool = False,
342 **kwargs: AttributeOverride,
343) -> SimpleStructureHook[Mapping[str, Any], T]:
344 """
345 Generate a specialized dict structuring function for a list of attributes.
347 Usually used as a building block by more specialized hook factories.
349 Any provided overrides are attached to the generated function under the
350 `overrides` attribute.
352 :param _cattrs_forbid_extra_keys: Whether the structuring function should raise a
353 `ForbiddenExtraKeysError` if unknown keys are encountered.
354 :param _cattrs_use_linecache: Whether to store the source code in the Python
355 linecache.
356 :param _cattrs_prefer_attrib_converters: If an _attrs_ converter is present on a
357 field, use it instead of processing the field normally.
358 :param _cattrs_detailed_validation: Whether to use a slower mode that produces
359 more detailed errors.
360 :param _cattrs_use_alias: If true, the attribute alias will be used as the
361 dictionary key by default.
362 :param _cattrs_include_init_false: If true, _attrs_ fields marked as `init=False`
363 will be included.
365 .. versionadded:: 24.1.0
366 .. versionchanged:: 25.2.0
367 The `_cattrs_use_alias` parameter takes its value from the given converter
368 by default.
369 .. versionchanged:: 26.1.0
370 `typing.Annotated[T, override()]` is now recognized and can be used to customize
371 unstructuring.
372 """
374 cl_name = cl.__name__
375 fn_name = "structure_" + cl_name
377 # We have generic parameters and need to generate a unique name for the function
378 for p in getattr(cl, "__parameters__", ()):
379 # This is nasty, I am not sure how best to handle `typing.List[str]` or
380 # `TClass[int, int]` as a parameter type here
381 try:
382 name_base = typevar_map[p.__name__]
383 except KeyError:
384 pn = p.__name__
385 raise StructureHandlerNotFoundError(
386 f"Missing type for generic argument {pn}, specify it when structuring.",
387 p,
388 ) from None
389 name = getattr(name_base, "__name__", None) or str(name_base)
390 # `<>` can be present in lambdas
391 # `|` can be present in unions
392 name = re.sub(r"[\[\.\] ,<>]", "_", name)
393 name = re.sub(r"\|", "u", name)
394 fn_name += f"_{name}"
396 internal_arg_parts = {"__cl": cl}
397 globs = {}
398 lines = []
399 post_lines = []
400 pi_lines = [] # post instantiation lines
401 invocation_lines = []
403 allowed_fields = set()
404 if _cattrs_forbid_extra_keys == "from_converter":
405 # BaseConverter doesn't have it so we're careful.
406 _cattrs_forbid_extra_keys = getattr(converter, "forbid_extra_keys", False)
407 if _cattrs_use_alias == "from_converter":
408 # BaseConverter doesn't have it so we're careful.
409 _cattrs_use_alias = getattr(converter, "use_alias", False)
410 if _cattrs_detailed_validation == "from_converter":
411 _cattrs_detailed_validation = converter.detailed_validation
412 if _cattrs_prefer_attrib_converters == "from_converter":
413 _cattrs_prefer_attrib_converters = converter._prefer_attrib_converters
415 if _cattrs_forbid_extra_keys:
416 globs["__c_a"] = allowed_fields
417 globs["__c_feke"] = ForbiddenExtraKeysError
419 if _cattrs_detailed_validation:
420 lines.append(" res = {}")
421 lines.append(" errors = []")
422 invocation_lines.append("**res,")
423 internal_arg_parts["__c_cve"] = ClassValidationError
424 internal_arg_parts["__c_avn"] = AttributeValidationNote
425 for a in attrs:
426 an = a.name
427 if an in kwargs:
428 override = kwargs[an]
429 else:
430 override = _annotated_override_or_default(a.type, neutral)
431 if override != neutral:
432 kwargs[an] = override
434 if override.omit:
435 continue
436 if override.omit is None and not a.init and not _cattrs_include_init_false:
437 continue
438 t = a.type
439 if isinstance(t, TypeVar):
440 t = typevar_map.get(t.__name__, t)
441 elif is_generic(t) and not is_bare(t) and not is_annotated(t):
442 t = deep_copy_with(t, typevar_map, cl)
444 # For each attribute, we try resolving the type here and now.
445 # If a type is manually overwritten, this function should be
446 # regenerated.
447 if override.struct_hook is not None:
448 # If the user has requested an override, just use that.
449 handler = override.struct_hook
450 else:
451 handler = find_structure_handler(
452 a, t, converter, _cattrs_prefer_attrib_converters
453 )
455 struct_handler_name = f"__c_structure_{an}"
456 if handler is not None:
457 internal_arg_parts[struct_handler_name] = handler
459 ian = a.alias
460 if override.rename is None:
461 kn = an if not _cattrs_use_alias else a.alias
462 if kn != an:
463 kwargs[an] = evolve(override, rename=kn)
464 else:
465 kn = override.rename
467 allowed_fields.add(kn)
468 i = " "
470 if not a.init:
471 if a.default is not NOTHING:
472 pi_lines.append(f"{i}if {kn!r} in o:")
473 i = f"{i} "
474 pi_lines.append(f"{i}try:")
475 i = f"{i} "
476 type_name = f"__c_type_{an}"
477 internal_arg_parts[type_name] = t
478 if handler is not None:
479 if handler == converter._structure_call:
480 internal_arg_parts[struct_handler_name] = t
481 pi_lines.append(
482 f"{i}instance.{an} = {struct_handler_name}(o[{kn!r}])"
483 )
484 else:
485 tn = f"__c_type_{an}"
486 internal_arg_parts[tn] = t
487 pi_lines.append(
488 f"{i}instance.{an} = {struct_handler_name}(o[{kn!r}], {tn})"
489 )
490 else:
491 pi_lines.append(f"{i}instance.{an} = o[{kn!r}]")
492 i = i[:-2]
493 pi_lines.append(f"{i}except Exception as e:")
494 i = f"{i} "
495 pi_lines.append(
496 f'{i}e.__notes__ = getattr(e, \'__notes__\', []) + [__c_avn("Structuring class {cl.__qualname__} @ attribute {an}", "{an}", __c_type_{an})]'
497 )
498 pi_lines.append(f"{i}errors.append(e)")
500 else:
501 if a.default is not NOTHING:
502 lines.append(f"{i}if {kn!r} in o:")
503 i = f"{i} "
504 lines.append(f"{i}try:")
505 i = f"{i} "
506 type_name = f"__c_type_{an}"
507 internal_arg_parts[type_name] = t
508 if handler:
509 if handler == converter._structure_call:
510 internal_arg_parts[struct_handler_name] = t
511 lines.append(
512 f"{i}res['{ian}'] = {struct_handler_name}(o[{kn!r}])"
513 )
514 else:
515 lines.append(
516 f"{i}res['{ian}'] = {struct_handler_name}(o[{kn!r}], {type_name})"
517 )
518 else:
519 lines.append(f"{i}res['{ian}'] = o[{kn!r}]")
520 i = i[:-2]
521 lines.append(f"{i}except Exception as e:")
522 i = f"{i} "
523 lines.append(
524 f'{i}e.__notes__ = getattr(e, \'__notes__\', []) + [__c_avn("Structuring class {cl.__qualname__} @ attribute {an}", "{an}", __c_type_{an})]'
525 )
526 lines.append(f"{i}errors.append(e)")
528 if _cattrs_forbid_extra_keys:
529 post_lines += [
530 " unknown_fields = set(o.keys()) - __c_a",
531 " if unknown_fields:",
532 " errors.append(__c_feke('', __cl, unknown_fields))",
533 ]
535 post_lines.append(
536 f" if errors: raise __c_cve('While structuring ' + {cl_name!r}, errors, __cl)"
537 )
538 if not pi_lines:
539 instantiation_lines = (
540 [" try:"]
541 + [" return __cl("]
542 + [f" {line}" for line in invocation_lines]
543 + [" )"]
544 + [
545 f" except Exception as exc: raise __c_cve('While structuring ' + {cl_name!r}, [exc], __cl)"
546 ]
547 )
548 else:
549 instantiation_lines = (
550 [" try:"]
551 + [" instance = __cl("]
552 + [f" {line}" for line in invocation_lines]
553 + [" )"]
554 + [
555 f" except Exception as exc: raise __c_cve('While structuring ' + {cl_name!r}, [exc], __cl)"
556 ]
557 )
558 pi_lines.append(" return instance")
559 else:
560 non_required = []
561 # The first loop deals with required args.
562 for a in attrs:
563 an = a.name
565 if an in kwargs:
566 override = kwargs[an]
567 else:
568 override = _annotated_override_or_default(a.type, neutral)
569 if override != neutral:
570 kwargs[an] = override
572 if override.omit:
573 continue
574 if override.omit is None and not a.init and not _cattrs_include_init_false:
575 continue
577 if a.default is not NOTHING:
578 non_required.append(a)
579 # The next loop will handle it.
580 continue
582 t = a.type
583 if isinstance(t, TypeVar):
584 t = typevar_map.get(t.__name__, t)
585 elif is_generic(t) and not is_bare(t) and not is_annotated(t):
586 t = deep_copy_with(t, typevar_map, cl)
588 # For each attribute, we try resolving the type here and now.
589 # If a type is manually overwritten, this function should be
590 # regenerated.
591 if override.struct_hook is not None:
592 # If the user has requested an override, just use that.
593 handler = override.struct_hook
594 else:
595 handler = find_structure_handler(
596 a, t, converter, _cattrs_prefer_attrib_converters
597 )
599 if override.rename is None:
600 kn = an if not _cattrs_use_alias else a.alias
601 if kn != an:
602 kwargs[an] = evolve(override, rename=kn)
603 else:
604 kn = override.rename
605 allowed_fields.add(kn)
607 if not a.init:
608 if handler is not None:
609 struct_handler_name = f"__c_structure_{an}"
610 internal_arg_parts[struct_handler_name] = handler
611 if handler == converter._structure_call:
612 internal_arg_parts[struct_handler_name] = t
613 pi_line = f" instance.{an} = {struct_handler_name}(o[{kn!r}])"
614 else:
615 tn = f"__c_type_{an}"
616 internal_arg_parts[tn] = t
617 pi_line = (
618 f" instance.{an} = {struct_handler_name}(o[{kn!r}], {tn})"
619 )
620 else:
621 pi_line = f" instance.{an} = o[{kn!r}]"
623 pi_lines.append(pi_line)
624 else:
625 if handler:
626 struct_handler_name = f"__c_structure_{an}"
627 internal_arg_parts[struct_handler_name] = handler
628 if handler == converter._structure_call:
629 internal_arg_parts[struct_handler_name] = t
630 invocation_line = f"{struct_handler_name}(o[{kn!r}]),"
631 else:
632 tn = f"__c_type_{an}"
633 internal_arg_parts[tn] = t
634 invocation_line = f"{struct_handler_name}(o[{kn!r}], {tn}),"
635 else:
636 invocation_line = f"o[{kn!r}],"
638 if a.kw_only:
639 invocation_line = f"{a.alias}={invocation_line}"
640 invocation_lines.append(invocation_line)
642 # The second loop is for optional args.
643 if non_required:
644 invocation_lines.append("**res,")
645 lines.append(" res = {}")
647 for a in non_required:
648 an = a.name
649 override = kwargs.get(an, neutral)
650 t = a.type
651 if isinstance(t, TypeVar):
652 t = typevar_map.get(t.__name__, t)
653 elif is_generic(t) and not is_bare(t) and not is_annotated(t):
654 t = deep_copy_with(t, typevar_map, cl)
656 # For each attribute, we try resolving the type here and now.
657 # If a type is manually overwritten, this function should be
658 # regenerated.
659 if override.struct_hook is not None:
660 # If the user has requested an override, just use that.
661 handler = override.struct_hook
662 else:
663 handler = find_structure_handler(
664 a, t, converter, _cattrs_prefer_attrib_converters
665 )
667 struct_handler_name = f"__c_structure_{an}"
668 internal_arg_parts[struct_handler_name] = handler
670 if override.rename is None:
671 kn = an if not _cattrs_use_alias else a.alias
672 if kn != an:
673 kwargs[an] = evolve(override, rename=kn)
674 else:
675 kn = override.rename
676 allowed_fields.add(kn)
677 if not a.init:
678 pi_lines.append(f" if {kn!r} in o:")
679 if handler:
680 if handler == converter._structure_call:
681 internal_arg_parts[struct_handler_name] = t
682 pi_lines.append(
683 f" instance.{an} = {struct_handler_name}(o[{kn!r}])"
684 )
685 else:
686 tn = f"__c_type_{an}"
687 internal_arg_parts[tn] = t
688 pi_lines.append(
689 f" instance.{an} = {struct_handler_name}(o[{kn!r}], {tn})"
690 )
691 else:
692 pi_lines.append(f" instance.{an} = o[{kn!r}]")
693 else:
694 post_lines.append(f" if {kn!r} in o:")
695 if handler:
696 if handler == converter._structure_call:
697 internal_arg_parts[struct_handler_name] = t
698 post_lines.append(
699 f" res['{a.alias}'] = {struct_handler_name}(o[{kn!r}])"
700 )
701 else:
702 tn = f"__c_type_{an}"
703 internal_arg_parts[tn] = t
704 post_lines.append(
705 f" res['{a.alias}'] = {struct_handler_name}(o[{kn!r}], {tn})"
706 )
707 else:
708 post_lines.append(f" res['{a.alias}'] = o[{kn!r}]")
709 if not pi_lines:
710 instantiation_lines = (
711 [" return __cl("]
712 + [f" {line}" for line in invocation_lines]
713 + [" )"]
714 )
715 else:
716 instantiation_lines = (
717 [" instance = __cl("]
718 + [f" {line}" for line in invocation_lines]
719 + [" )"]
720 )
721 pi_lines.append(" return instance")
723 if _cattrs_forbid_extra_keys:
724 post_lines += [
725 " unknown_fields = set(o.keys()) - __c_a",
726 " if unknown_fields:",
727 " raise __c_feke('', __cl, unknown_fields)",
728 ]
730 # At the end, we create the function header.
731 internal_arg_line = ", ".join([f"{i}={i}" for i in internal_arg_parts])
732 globs.update(internal_arg_parts)
734 total_lines = [
735 f"def {fn_name}(o, _=__cl, {internal_arg_line}):",
736 *lines,
737 *post_lines,
738 *instantiation_lines,
739 *pi_lines,
740 ]
742 script = "\n".join(total_lines)
743 fname = generate_unique_filename(
744 cl, "structure", lines=total_lines if _cattrs_use_linecache else []
745 )
747 eval(compile(script, fname, "exec"), globs)
749 res = globs[fn_name]
750 res.overrides = kwargs
752 return res
755def make_dict_structure_fn(
756 cl: type[T],
757 converter: BaseConverter,
758 _cattrs_forbid_extra_keys: bool | Literal["from_converter"] = "from_converter",
759 _cattrs_use_linecache: bool = True,
760 _cattrs_prefer_attrib_converters: (
761 bool | Literal["from_converter"]
762 ) = "from_converter",
763 _cattrs_detailed_validation: bool | Literal["from_converter"] = "from_converter",
764 _cattrs_use_alias: bool | Literal["from_converter"] = "from_converter",
765 _cattrs_include_init_false: bool = False,
766 **kwargs: AttributeOverride,
767) -> SimpleStructureHook[Mapping[str, Any], T]:
768 """
769 Generate a specialized dict structuring function for an attrs class or
770 dataclass.
772 Any provided overrides are attached to the generated function under the
773 `overrides` attribute.
775 :param _cattrs_forbid_extra_keys: Whether the structuring function should raise a
776 `ForbiddenExtraKeysError` if unknown keys are encountered.
777 :param _cattrs_use_linecache: Whether to store the source code in the Python
778 linecache.
779 :param _cattrs_prefer_attrib_converters: If an _attrs_ converter is present on a
780 field, use it instead of processing the field normally.
781 :param _cattrs_detailed_validation: Whether to use a slower mode that produces
782 more detailed errors.
783 :param _cattrs_use_alias: If true, the attribute alias will be used as the
784 dictionary key by default.
785 :param _cattrs_include_init_false: If true, _attrs_ fields marked as `init=False`
786 will be included.
788 .. versionadded:: 23.2.0 *_cattrs_use_alias*
789 .. versionadded:: 23.2.0 *_cattrs_include_init_false*
790 .. versionchanged:: 23.2.0
791 The `_cattrs_forbid_extra_keys` and `_cattrs_detailed_validation` parameters
792 take their values from the given converter by default.
793 .. versionchanged:: 24.1.0
794 The `_cattrs_prefer_attrib_converters` parameter takes its value from the given
795 converter by default.
796 .. versionchanged:: 25.2.0
797 The `_cattrs_use_alias` parameter takes its value from the given converter
798 by default.
799 .. versionchanged:: 26.1.0
800 `typing.Annotated[T, override()]` is now recognized and can be used to customize
801 unstructuring.
802 """
804 mapping = {}
805 if is_generic(cl):
806 base = get_origin(cl)
807 mapping = generate_mapping(cl, mapping)
808 if base is not None:
809 cl = base
811 for base in getattr(cl, "__orig_bases__", ()):
812 if is_generic(base) and not str(base).startswith("typing.Generic"):
813 mapping = generate_mapping(base, mapping)
814 break
816 attrs = adapted_fields(cl)
818 # We keep track of what we're generating to help with recursive
819 # class graphs.
820 try:
821 working_set = already_generating.working_set
822 except AttributeError:
823 working_set = set()
824 already_generating.working_set = working_set
825 else:
826 if cl in working_set:
827 raise RecursionError()
829 working_set.add(cl)
831 try:
832 return make_dict_structure_fn_from_attrs(
833 attrs,
834 cl,
835 converter,
836 mapping,
837 _cattrs_forbid_extra_keys=_cattrs_forbid_extra_keys,
838 _cattrs_use_linecache=_cattrs_use_linecache,
839 _cattrs_prefer_attrib_converters=_cattrs_prefer_attrib_converters,
840 _cattrs_detailed_validation=_cattrs_detailed_validation,
841 _cattrs_use_alias=_cattrs_use_alias,
842 _cattrs_include_init_false=_cattrs_include_init_false,
843 **kwargs,
844 )
845 finally:
846 working_set.remove(cl)
847 if not working_set:
848 del already_generating.working_set
851IterableUnstructureFn = Callable[[Iterable[Any]], Any]
854#: A type alias for heterogeneous tuple unstructure hooks.
855HeteroTupleUnstructureFn: TypeAlias = Callable[[tuple[Any, ...]], Any]
856HeteroTupleStructureFn: TypeAlias = Callable[[Iterable[Any], Any], tuple[Any, ...]]
859def make_hetero_tuple_structure_fn(
860 cl: Any,
861 converter: BaseConverter,
862 detailed_validation: bool | Literal["from_converter"] = "from_converter",
863 use_linecache: bool = True,
864) -> HeteroTupleStructureFn:
865 """Generate a specialized structuring function for a heterogeneous tuple.
867 .. versionadded:: 26.2.0
868 """
869 fn_name = "structure_tuple"
871 if detailed_validation == "from_converter":
872 detailed_validation = converter.detailed_validation
874 type_args = get_args(cl)
875 globs = {}
876 lines = []
877 internal_arg_parts = {"__cl": cl}
879 if detailed_validation:
880 internal_arg_parts["__c_ive"] = IterableValidationError
881 internal_arg_parts["__c_ivn"] = IterableValidationNote
882 lines.extend([" errors = []", " res = []"])
884 for ix, t in enumerate(type_args):
885 handler = converter.get_structure_hook(t)
886 struct_handler_name = f"__c_structure_{ix}"
887 type_name = f"__c_type_{ix}"
888 internal_arg_parts[struct_handler_name] = handler
889 internal_arg_parts[type_name] = t
890 if handler == converter._structure_call:
891 internal_arg_parts[struct_handler_name] = t
892 invocation = f"{struct_handler_name}(o[{ix}])"
893 else:
894 invocation = f"{struct_handler_name}(o[{ix}], {type_name})"
895 note = f"Structuring {cl} @ index {ix}"
896 lines.extend(
897 [
898 f" if len(o) > {ix}:",
899 " try:",
900 f" res.append({invocation})",
901 " except Exception as e:",
902 (
903 f" e.__notes__ = [*getattr(e, '__notes__', []), "
904 f"__c_ivn({note!r}, {ix}, {type_name})]"
905 ),
906 " errors.append(e)",
907 ]
908 )
910 lines.extend(
911 [
912 f" if len(o) != {len(type_args)}:",
913 (
914 " problem = 'Not enough' if len(o) < "
915 f"{len(type_args)} else 'Too many'"
916 ),
917 ' exc = ValueError(f"{problem} values in {o!r} to structure as {__cl!r}")',
918 " exc.__notes__ = [f'Structuring {__cl}']",
919 " errors.append(exc)",
920 " if errors:",
921 " raise __c_ive(f'While structuring {__cl!r}', errors, __cl)",
922 " return tuple(res)",
923 ]
924 )
925 else:
926 for ix, t in enumerate(type_args):
927 handler = converter.get_structure_hook(t)
928 struct_handler_name = f"__c_structure_{ix}"
929 type_name = f"__c_type_{ix}"
930 internal_arg_parts[struct_handler_name] = handler
931 internal_arg_parts[type_name] = t
932 if handler == converter._structure_call:
933 internal_arg_parts[struct_handler_name] = t
934 invocation = f"{struct_handler_name}(o[{ix}])"
935 else:
936 invocation = f"{struct_handler_name}(o[{ix}], {type_name})"
937 lines.append(f" {invocation},")
939 total_len = len(type_args)
940 lines = [
941 f" if len(o) != {total_len}:",
942 (f" problem = 'Not enough' if len(o) < {total_len} else 'Too many'"),
943 ' raise ValueError(f"{problem} values in {o!r} to structure as {__cl!r}")',
944 " return (",
945 *lines,
946 " )",
947 ]
949 internal_arg_line = ", ".join([f"{i}={i}" for i in internal_arg_parts])
950 globs.update(internal_arg_parts)
951 script = "\n".join([f"def {fn_name}(o, _=__cl, {internal_arg_line}):", *lines])
953 fname = generate_unique_filename(
954 cl, "structure", lines=script.splitlines() if use_linecache else []
955 )
956 eval(compile(script, fname, "exec"), globs)
957 return globs[fn_name]
960def make_hetero_tuple_unstructure_fn(
961 cl: Any,
962 converter: BaseConverter,
963 unstructure_to: Any = None,
964 type_args: tuple | None = None,
965) -> HeteroTupleUnstructureFn:
966 """Generate a specialized unstructure function for a heterogeneous tuple.
968 :param type_args: If provided, override the type arguments.
969 """
970 fn_name = "unstructure_tuple"
972 type_args = get_args(cl) if type_args is None else type_args
974 # We can do the dispatch here and now.
975 handlers = [converter.get_unstructure_hook(type_arg) for type_arg in type_args]
977 globs = {f"__cattr_u_{i}": h for i, h in enumerate(handlers)}
978 if unstructure_to is not tuple:
979 globs["__cattr_seq_cl"] = unstructure_to or cl
980 lines = []
982 lines.append(f"def {fn_name}(tup):")
983 if unstructure_to is not tuple:
984 lines.append(" res = __cattr_seq_cl((")
985 else:
986 lines.append(" res = (")
987 for i in range(len(handlers)):
988 if handlers[i] == identity:
989 lines.append(f" tup[{i}],")
990 else:
991 lines.append(f" __cattr_u_{i}(tup[{i}]),")
993 if unstructure_to is not tuple:
994 lines.append(" ))")
995 else:
996 lines.append(" )")
998 total_lines = [*lines, " return res"]
1000 eval(compile("\n".join(total_lines), "", "exec"), globs)
1002 return globs[fn_name]
1005MappingUnstructureFn = Callable[[Mapping[Any, Any]], Any]
1008# This factory is here for backwards compatibility and circular imports.
1009def mapping_unstructure_factory(
1010 cl: Any,
1011 converter: BaseConverter,
1012 unstructure_to: Any = None,
1013 key_handler: Callable[[Any, Any | None], Any] | None = None,
1014) -> MappingUnstructureFn:
1015 """Generate a specialized unstructure function for a mapping.
1017 :param unstructure_to: The class to unstructure to; defaults to the
1018 same class as the mapping being unstructured.
1019 """
1020 kh = key_handler or converter.unstructure
1021 val_handler = converter.unstructure
1023 fn_name = "unstructure_mapping"
1024 origin = cl
1026 # Let's try fishing out the type args.
1027 if getattr(cl, "__args__", None) is not None:
1028 args = get_args(cl)
1029 if len(args) == 2:
1030 key_arg, val_arg = args
1031 else:
1032 # Probably a Counter
1033 key_arg, val_arg = args[0], Any
1034 # We can do the dispatch here and now.
1035 kh = key_handler or converter.get_unstructure_hook(key_arg, cache_result=False)
1036 if kh == identity:
1037 kh = None
1039 val_handler = converter.get_unstructure_hook(val_arg, cache_result=False)
1040 if val_handler == identity:
1041 val_handler = None
1043 origin = get_origin(cl)
1045 globs = {"__cattr_k_u": kh, "__cattr_v_u": val_handler}
1047 k_u = "__cattr_k_u(k)" if kh is not None else "k"
1048 v_u = "__cattr_v_u(v)" if val_handler is not None else "v"
1050 lines = [f"def {fn_name}(mapping):"]
1052 if unstructure_to is dict or (unstructure_to is None and origin is dict):
1053 if kh is None and val_handler is None:
1054 # Simplest path.
1055 return dict
1057 lines.append(f" return {{{k_u}: {v_u} for k, v in mapping.items()}}")
1058 else:
1059 globs["__cattr_mapping_cl"] = unstructure_to or cl
1060 lines.append(
1061 f" res = __cattr_mapping_cl(({k_u}, {v_u}) for k, v in mapping.items())"
1062 )
1064 lines = [*lines, " return res"]
1066 eval(compile("\n".join(lines), "", "exec"), globs)
1068 return globs[fn_name]
1071make_mapping_unstructure_fn: Final = mapping_unstructure_factory
1074# This factory is here for backwards compatibility and circular imports.
1075def mapping_structure_factory(
1076 cl: type[T],
1077 converter: BaseConverter,
1078 structure_to: type = dict,
1079 key_type=NOTHING,
1080 val_type=NOTHING,
1081 detailed_validation: bool | Literal["from_converter"] = "from_converter",
1082) -> SimpleStructureHook[Mapping[Any, Any], T]:
1083 """Generate a specialized structure function for a mapping."""
1084 fn_name = "structure_mapping"
1086 if detailed_validation == "from_converter":
1087 detailed_validation = converter.detailed_validation
1089 globs: dict[str, type] = {"__cattr_mapping_cl": structure_to}
1091 lines = []
1092 internal_arg_parts = {}
1094 # Let's try fishing out the type args.
1095 if not is_bare(cl):
1096 args = get_args(cl)
1097 if len(args) == 2:
1098 key_arg_cand, val_arg_cand = args
1099 if key_type is NOTHING:
1100 key_type = key_arg_cand
1101 if val_type is NOTHING:
1102 val_type = val_arg_cand
1103 else:
1104 if key_type is not NOTHING and val_type is NOTHING:
1105 (val_type,) = args
1106 elif key_type is NOTHING and val_type is not NOTHING:
1107 (key_type,) = args
1108 else:
1109 # Probably a Counter
1110 (key_type,) = args
1111 val_type = Any
1113 is_bare_dict = val_type in ANIES and key_type in ANIES
1114 if not is_bare_dict:
1115 # We can do the dispatch here and now.
1116 key_handler = converter.get_structure_hook(key_type, cache_result=False)
1117 if key_handler == converter._structure_call:
1118 key_handler = key_type
1120 val_handler = converter.get_structure_hook(val_type, cache_result=False)
1121 if val_handler == converter._structure_call:
1122 val_handler = val_type
1124 globs["__cattr_k_t"] = key_type
1125 globs["__cattr_v_t"] = val_type
1126 globs["__cattr_k_s"] = key_handler
1127 globs["__cattr_v_s"] = val_handler
1128 k_s = (
1129 "__cattr_k_s(k, __cattr_k_t)"
1130 if key_handler != key_type
1131 else "__cattr_k_s(k)"
1132 )
1133 v_s = (
1134 "__cattr_v_s(v, __cattr_v_t)"
1135 if val_handler != val_type
1136 else "__cattr_v_s(v)"
1137 )
1138 else:
1139 is_bare_dict = True
1141 if is_bare_dict:
1142 # No args, it's a bare dict.
1143 lines.append(" res = dict(mapping)")
1144 else:
1145 if detailed_validation:
1146 internal_arg_parts["IterableValidationError"] = IterableValidationError
1147 internal_arg_parts["IterableValidationNote"] = IterableValidationNote
1148 internal_arg_parts["val_type"] = (
1149 val_type if val_type is not NOTHING else Any
1150 )
1151 internal_arg_parts["key_type"] = (
1152 key_type if key_type is not NOTHING else Any
1153 )
1154 globs["enumerate"] = enumerate
1156 lines.append(" res = {}; errors = []")
1157 lines.append(" for k, v in mapping.items():")
1158 lines.append(" try:")
1159 lines.append(f" value = {v_s}")
1160 lines.append(" except Exception as e:")
1161 lines.append(
1162 " e.__notes__ = getattr(e, '__notes__', []) + [IterableValidationNote(f'Structuring mapping value @ key {k!r}', k, val_type)]"
1163 )
1164 lines.append(" errors.append(e)")
1165 lines.append(" continue")
1166 lines.append(" try:")
1167 lines.append(f" key = {k_s}")
1168 lines.append(" res[key] = value")
1169 lines.append(" except Exception as e:")
1170 lines.append(
1171 " e.__notes__ = getattr(e, '__notes__', []) + [IterableValidationNote(f'Structuring mapping key @ key {k!r}', k, key_type)]"
1172 )
1173 lines.append(" errors.append(e)")
1174 lines.append(" if errors:")
1175 lines.append(
1176 f" raise IterableValidationError('While structuring ' + {repr(cl)!r}, errors, __cattr_mapping_cl)"
1177 )
1178 else:
1179 lines.append(f" res = {{{k_s}: {v_s} for k, v in mapping.items()}}")
1180 if structure_to is not dict:
1181 lines.append(" res = __cattr_mapping_cl(res)")
1183 internal_arg_line = ", ".join([f"{i}={i}" for i in internal_arg_parts])
1184 if internal_arg_line:
1185 internal_arg_line = f", {internal_arg_line}"
1186 for k, v in internal_arg_parts.items():
1187 globs[k] = v
1189 globs["cl"] = cl
1190 def_line = f"def {fn_name}(mapping, cl=cl{internal_arg_line}):"
1191 total_lines = [def_line, *lines, " return res"]
1192 script = "\n".join(total_lines)
1194 eval(compile(script, "", "exec"), globs)
1196 return globs[fn_name]
1199make_mapping_structure_fn: Final = mapping_structure_factory
1202# This factory is here for backwards compatibility and circular imports.
1203def iterable_unstructure_factory(
1204 cl: Any, converter: BaseConverter, unstructure_to: Any = None
1205) -> UnstructureHook:
1206 """A hook factory for unstructuring iterables.
1208 :param unstructure_to: Force unstructuring to this type, if provided.
1210 .. versionchanged:: 24.2.0
1211 `typing.NoDefault` is now correctly handled as `Any`.
1212 """
1213 handler = converter.unstructure
1215 # Let's try fishing out the type args
1216 # Unspecified tuples have `__args__` as empty tuples, so guard
1217 # against IndexError.
1218 if getattr(cl, "__args__", None) not in (None, ()):
1219 type_arg = cl.__args__[0]
1220 if isinstance(type_arg, TypeVar):
1221 type_arg = getattr(type_arg, "__default__", Any)
1222 if type_arg is NoDefault:
1223 type_arg = Any
1224 handler = converter.get_unstructure_hook(type_arg, cache_result=False)
1225 if handler == identity:
1226 # Save ourselves the trouble of iterating over it all.
1227 return unstructure_to or cl
1229 def unstructure_iterable(iterable, _seq_cl=unstructure_to or cl, _hook=handler):
1230 return _seq_cl(_hook(i) for i in iterable)
1232 return unstructure_iterable
1235make_iterable_unstructure_fn: Final = iterable_unstructure_factory