Skip to content

union_field

ExceptionGroupError

Bases: MarshmallowUnionError

Collection of possibly multiple exceptions.

Source code in griptape/schemas/union_field.py 
14
15
16
17
18
19
20
class ExceptionGroupError(MarshmallowUnionError):
    """Collection of possibly multiple exceptions."""

    def __init__(self, msg: str, errors: Any) -> None:
        self.msg = msg
        self.errors = errors
        super().__init__(msg, errors)

errors = errors instance-attribute

msg = msg instance-attribute

__init__(msg, errors)

Source code in griptape/schemas/union_field.py 
17
18
19
20
def __init__(self, msg: str, errors: Any) -> None:
    self.msg = msg
    self.errors = errors
    super().__init__(msg, errors)

MarshmallowUnionError

Bases: Exception

Base error for marshmallow_union.

Source code in griptape/schemas/union_field.py 
10
11
class MarshmallowUnionError(Exception):
    """Base error for marshmallow_union."""

Union

Bases: Field

Field that accepts any one of multiple fields.

Source: https://github.com/adamboche/python-marshmallow-union Each argument will be tried until one succeeds.

Parameters:

Name Type Description Default
fields list[Field]

The list of candidate fields to try.

 required
reverse_serialize_candidates bool

Whether to try the candidates in reverse order when serializing.

 False
Source code in griptape/schemas/union_field.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class Union(marshmallow.fields.Field):
    """Field that accepts any one of multiple fields.

    Source: https://github.com/adamboche/python-marshmallow-union
    Each argument will be tried until one succeeds.

    Args:
        fields: The list of candidate fields to try.
        reverse_serialize_candidates: Whether to try the candidates in reverse order when serializing.
    """

    def __init__(
        self,
        fields: list[marshmallow.fields.Field],
        *,
        reverse_serialize_candidates: bool = False,
        **kwargs: Any,
    ) -> None:
        self._candidate_fields = fields
        self._reverse_serialize_candidates = reverse_serialize_candidates
        super().__init__(**kwargs)

    def _serialize(self, value: Any, attr: str | None, obj: str, **kwargs: Any) -> Any:
        """Pulls the value for the given key from the object, applies the field's formatting and returns the result.

        Args:
            value: The value to be serialized.
            attr: The attribute or key to get from the object.
            obj: The object to pull the key from.
            kwargs: Field-specific keyword arguments.

        Raises:
            marshmallow.exceptions.ValidationError: In case of formatting problem
        """
        error_store = kwargs.pop("error_store", marshmallow.error_store.ErrorStore())
        fields = (
            list(reversed(self._candidate_fields)) if self._reverse_serialize_candidates else self._candidate_fields
        )

        for candidate_field in fields:
            try:
                # pylint: disable=protected-access
                return candidate_field._serialize(value, attr, obj, error_store=error_store, **kwargs)
            except (TypeError, ValueError) as e:
                error_store.store_error({attr: str(e)})

        raise ExceptionGroupError("All serializers raised exceptions.", error_store.errors)

    def _deserialize(self, value: Any, attr: str | None = None, data: Any = None, **kwargs: Any) -> Any:
        """Deserialize ``value``.

        Args:
            value: The value to be deserialized.
            attr: The attribute/key in `data` to be deserialized.
            data: The raw input data passed to the `Schema.load`.
            kwargs: Field-specific keyword arguments.

        Raises:
            ValidationError: If an invalid value is passed or if a required value is missing.
        """
        errors = []
        for candidate_field in self._candidate_fields:
            try:
                return candidate_field.deserialize(value, attr, data, **kwargs)
            except marshmallow.exceptions.ValidationError as exc:
                errors.append(exc.messages)

        raise marshmallow.exceptions.ValidationError(message=errors, field_name=attr or "")

_candidate_fields = fields instance-attribute

_reverse_serialize_candidates = reverse_serialize_candidates instance-attribute

__init__(fields, *, reverse_serialize_candidates=False, **kwargs)

Source code in griptape/schemas/union_field.py 
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    fields: list[marshmallow.fields.Field],
    *,
    reverse_serialize_candidates: bool = False,
    **kwargs: Any,
) -> None:
    self._candidate_fields = fields
    self._reverse_serialize_candidates = reverse_serialize_candidates
    super().__init__(**kwargs)

_deserialize(value, attr=None, data=None, **kwargs)

Deserialize value.

Parameters:

Name Type Description Default
value Any

The value to be deserialized.

 required
attr str | None

The attribute/key in data to be deserialized.

 None
data Any

The raw input data passed to the Schema.load.

 None
kwargs Any

Field-specific keyword arguments.

 {}

Raises:

Type Description
ValidationError

If an invalid value is passed or if a required value is missing.
Source code in griptape/schemas/union_field.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def _deserialize(self, value: Any, attr: str | None = None, data: Any = None, **kwargs: Any) -> Any:
    """Deserialize ``value``.

    Args:
        value: The value to be deserialized.
        attr: The attribute/key in `data` to be deserialized.
        data: The raw input data passed to the `Schema.load`.
        kwargs: Field-specific keyword arguments.

    Raises:
        ValidationError: If an invalid value is passed or if a required value is missing.
    """
    errors = []
    for candidate_field in self._candidate_fields:
        try:
            return candidate_field.deserialize(value, attr, data, **kwargs)
        except marshmallow.exceptions.ValidationError as exc:
            errors.append(exc.messages)

    raise marshmallow.exceptions.ValidationError(message=errors, field_name=attr or "")

_serialize(value, attr, obj, **kwargs)

Pulls the value for the given key from the object, applies the field's formatting and returns the result.

Parameters:

Name Type Description Default
value Any

The value to be serialized.

 required
attr str | None

The attribute or key to get from the object.

 required
obj str

The object to pull the key from.

 required
kwargs Any

Field-specific keyword arguments.

 {}

Raises:

Type Description
ValidationError

In case of formatting problem
Source code in griptape/schemas/union_field.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def _serialize(self, value: Any, attr: str | None, obj: str, **kwargs: Any) -> Any:
    """Pulls the value for the given key from the object, applies the field's formatting and returns the result.

    Args:
        value: The value to be serialized.
        attr: The attribute or key to get from the object.
        obj: The object to pull the key from.
        kwargs: Field-specific keyword arguments.

    Raises:
        marshmallow.exceptions.ValidationError: In case of formatting problem
    """
    error_store = kwargs.pop("error_store", marshmallow.error_store.ErrorStore())
    fields = (
        list(reversed(self._candidate_fields)) if self._reverse_serialize_candidates else self._candidate_fields
    )

    for candidate_field in fields:
        try:
            # pylint: disable=protected-access
            return candidate_field._serialize(value, attr, obj, error_store=error_store, **kwargs)
        except (TypeError, ValueError) as e:
            error_store.store_error({attr: str(e)})

    raise ExceptionGroupError("All serializers raised exceptions.", error_store.errors)