Skip to content

wg_utilities.helpers

Sentinel

Dummy value for tripping conditions, breaking loops, all sorts!

Source code in wg_utilities/helpers/__init__.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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
class Sentinel:
    """Dummy value for tripping conditions, breaking loops, all sorts!"""

    def __bool__(self) -> Literal[False]:
        """Always False, so that it can be used as a sentinel value."""
        return False

    def __call__(self, *_: Any, **__: Any) -> None:
        """Do nothing when called."""

    def __eq__(self, other: Any) -> bool:
        """Return True if the other value is also a Sentinel."""
        return isinstance(other, Sentinel)

    def __hash__(self) -> int:
        """Return a hash of the class name."""
        return hash(self.__class__.__name__)

    def __iter__(self) -> Iterator:
        """Return an iterator that will always raise StopIteration."""
        return self.Iterator()

    def __ne__(self, other: Any) -> bool:
        """Return False if the other value is also a Sentinel."""
        return not isinstance(other, Sentinel)

    def __repr__(self) -> str:
        """Return the class name."""
        return self.__class__.__name__

    class Iterator:
        """An iterator that will always raise StopIteration."""

        def __iter__(self) -> Sentinel.Iterator:
            """Return an iterator that will always raise StopIteration."""
            return self

        def __next__(self) -> Never:
            """Always raise StopIteration."""
            raise StopIteration

Iterator

An iterator that will always raise StopIteration.

Source code in wg_utilities/helpers/__init__.py 
38
39
40
41
42
43
44
45
46
47
class Iterator:
    """An iterator that will always raise StopIteration."""

    def __iter__(self) -> Sentinel.Iterator:
        """Return an iterator that will always raise StopIteration."""
        return self

    def __next__(self) -> Never:
        """Always raise StopIteration."""
        raise StopIteration

__iter__()

Return an iterator that will always raise StopIteration.

Source code in wg_utilities/helpers/__init__.py 
41
42
43
def __iter__(self) -> Sentinel.Iterator:
    """Return an iterator that will always raise StopIteration."""
    return self

__next__()

Always raise StopIteration.

Source code in wg_utilities/helpers/__init__.py 
45
46
47
def __next__(self) -> Never:
    """Always raise StopIteration."""
    raise StopIteration

__bool__()

Always False, so that it can be used as a sentinel value.

Source code in wg_utilities/helpers/__init__.py 
11
12
13
def __bool__(self) -> Literal[False]:
    """Always False, so that it can be used as a sentinel value."""
    return False

__call__(*_, **__)

Do nothing when called.

Source code in wg_utilities/helpers/__init__.py 
15
16
def __call__(self, *_: Any, **__: Any) -> None:
    """Do nothing when called."""

__eq__(other)

Return True if the other value is also a Sentinel.

Source code in wg_utilities/helpers/__init__.py 
18
19
20
def __eq__(self, other: Any) -> bool:
    """Return True if the other value is also a Sentinel."""
    return isinstance(other, Sentinel)

__hash__()

Return a hash of the class name.

Source code in wg_utilities/helpers/__init__.py 
22
23
24
def __hash__(self) -> int:
    """Return a hash of the class name."""
    return hash(self.__class__.__name__)

__iter__()

Return an iterator that will always raise StopIteration.

Source code in wg_utilities/helpers/__init__.py 
26
27
28
def __iter__(self) -> Iterator:
    """Return an iterator that will always raise StopIteration."""
    return self.Iterator()

__ne__(other)

Return False if the other value is also a Sentinel.

Source code in wg_utilities/helpers/__init__.py 
30
31
32
def __ne__(self, other: Any) -> bool:
    """Return False if the other value is also a Sentinel."""
    return not isinstance(other, Sentinel)

__repr__()

Return the class name.

Source code in wg_utilities/helpers/__init__.py 
34
35
36
def __repr__(self) -> str:
    """Return the class name."""
    return self.__class__.__name__

meta

PostInitMeta

Bases: type

Metaclass to call a post-init method after the class is instantiated.

Source code in wg_utilities/helpers/meta/post_init.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class PostInitMeta(type):
    """Metaclass to call a post-init method after the class is instantiated."""

    def __call__(cls, *args: Any, **kwargs: Any) -> Any:
        """Call the class and then call the post-init method."""
        obj = type.__call__(cls, *args, **kwargs)

        try:
            obj.__post_init__()
        except AttributeError as exc:
            if exc.name == "__post_init__":
                raise MissingPostInitMethodError(cls) from None

            raise

        return obj

__call__(*args, **kwargs)

Call the class and then call the post-init method.

Source code in wg_utilities/helpers/meta/post_init.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def __call__(cls, *args: Any, **kwargs: Any) -> Any:
    """Call the class and then call the post-init method."""
    obj = type.__call__(cls, *args, **kwargs)

    try:
        obj.__post_init__()
    except AttributeError as exc:
        if exc.name == "__post_init__":
            raise MissingPostInitMethodError(cls) from None

        raise

    return obj

post_init

Mixin to provide __post_init__ functionality.

MissingPostInitMethodError

Bases: BadDefinitionError

Raised when a class using the PostInitMeta metaclass does not define a __post_init__ method.

Source code in wg_utilities/helpers/meta/post_init.py 
10
11
12
13
14
15
class MissingPostInitMethodError(BadDefinitionError):
    """Raised when a class using the `PostInitMeta` metaclass does not define a `__post_init__` method."""

    def __init__(self, cls: type) -> None:
        """Initialize the error with the class that is missing the `__post_init__` method."""
        super().__init__(f"Class {cls.__name__!r} is missing a `__post_init__` method.")
__init__(cls)

Initialize the error with the class that is missing the __post_init__ method.

Source code in wg_utilities/helpers/meta/post_init.py 
13
14
15
def __init__(self, cls: type) -> None:
    """Initialize the error with the class that is missing the `__post_init__` method."""
    super().__init__(f"Class {cls.__name__!r} is missing a `__post_init__` method.")

PostInitMeta

Bases: type

Metaclass to call a post-init method after the class is instantiated.

Source code in wg_utilities/helpers/meta/post_init.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class PostInitMeta(type):
    """Metaclass to call a post-init method after the class is instantiated."""

    def __call__(cls, *args: Any, **kwargs: Any) -> Any:
        """Call the class and then call the post-init method."""
        obj = type.__call__(cls, *args, **kwargs)

        try:
            obj.__post_init__()
        except AttributeError as exc:
            if exc.name == "__post_init__":
                raise MissingPostInitMethodError(cls) from None

            raise

        return obj
__call__(*args, **kwargs)

Call the class and then call the post-init method.

Source code in wg_utilities/helpers/meta/post_init.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def __call__(cls, *args: Any, **kwargs: Any) -> Any:
    """Call the class and then call the post-init method."""
    obj = type.__call__(cls, *args, **kwargs)

    try:
        obj.__post_init__()
    except AttributeError as exc:
        if exc.name == "__post_init__":
            raise MissingPostInitMethodError(cls) from None

        raise

    return obj

mixin

InstanceCache

Mixin class to provide instance caching functionality.

Attributes:

Name Type Description
cache_id_attr str | None

The name of the attribute to use in the cache ID. Defaults to None. Must be provided if cache_id_func is None.
cache_id_func str | None

The name of the function whose return value to use in the cache ID. Defaults to "hash". Must not be None if cache_id_attr is None.
kwargs Any

Additional keyword arguments to pass to the superclass.
Source code in wg_utilities/helpers/mixin/instance_cache.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class InstanceCacheMixin(metaclass=PostInitMeta):
    """Mixin class to provide instance caching functionality.

    Attributes:
        cache_id_attr (str | None): The name of the attribute to use in the cache ID. Defaults to None.
            Must be provided if `cache_id_func` is None.
        cache_id_func (str | None): The name of the function whose return value to use in the cache ID.
            Defaults to "__hash__". Must not be `None` if `cache_id_attr` is None.
        kwargs (Any): Additional keyword arguments to pass to the superclass.
    """

    _INSTANCES: dict[CacheIdType, Self]

    _CACHE_ID_ATTR: ClassVar[str]
    _CACHE_ID_FUNC: ClassVar[str]

    def __init_subclass__(
        cls,
        *,
        cache_id_attr: str | None = None,
        cache_id_func: str | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialise subclasses with instance cache and cache ID attributes.

        Args:
            cache_id_attr (str | None, optional): The name of the attribute to use in the cache ID. Defaults to
                None. Must be provided if `cache_id_func` is None.
            cache_id_func (str | None, optional): The name of the function whose return value to use in the cache
                ID. Defaults to "__hash__" (and must be provided) if `cache_id_attr` is None.
            kwargs (Any): Additional keyword arguments to pass to the superclass.

        Raises:
            InstanceCacheSubclassError: If neither `cache_id_attr` nor `cache_id_func` are provided.
        """
        if not (cache_id_attr or cache_id_func):
            cache_id_func = "__hash__"

        if not (isinstance(cache_id_attr, str) or cache_id_attr is None):
            raise InstanceCacheSubclassError(
                f"Invalid cache ID attribute: {cache_id_attr!r}",
            )

        if not (isinstance(cache_id_func, str) or cache_id_func is None):
            raise InstanceCacheSubclassError(
                f"Invalid cache ID function: {cache_id_func!r}",
            )

        super().__init_subclass__(**kwargs)
        cls._INSTANCES = {}

        if cache_id_attr:
            cls._CACHE_ID_ATTR = cache_id_attr

        if cache_id_func:
            cls._CACHE_ID_FUNC = cache_id_func

    def __post_init__(self) -> None:
        """Add the instance to the cache.

        Raises:
            InstanceCacheDuplicateError: If the instance already exists in the cache.
        """
        if (instance_id := cache_id(self)) in self._INSTANCES:
            raise InstanceCacheDuplicateError(self.__class__, instance_id)

        self._INSTANCES[instance_id] = self  # type: ignore[assignment]

    @final
    @classmethod
    def from_cache(cls, cache_id_: CacheIdType, /) -> Self:
        """Get an instance from the cache by its cache ID."""
        try:
            return cls._INSTANCES[cache_id_]  # type: ignore[return-value]
        except KeyError:
            raise CacheIdNotFoundError(cls, cache_id_) from None

    @final
    @classmethod
    def has_cache_entry(cls, cache_id_: CacheIdType, /) -> bool:
        """Check if the cache has an entry for the given cache ID."""
        return cache_id_ in cls._INSTANCES

__init_subclass__(*, cache_id_attr=None, cache_id_func=None, **kwargs)

Initialise subclasses with instance cache and cache ID attributes.

Parameters:

Name Type Description Default
cache_id_attr str | None

The name of the attribute to use in the cache ID. Defaults to None. Must be provided if cache_id_func is None.

 None
cache_id_func str | None

The name of the function whose return value to use in the cache ID. Defaults to "hash" (and must be provided) if cache_id_attr is None.

 None
kwargs Any

Additional keyword arguments to pass to the superclass.

 {}

Raises:

Type Description
InstanceCacheSubclassError

If neither cache_id_attr nor cache_id_func are provided.
Source code in wg_utilities/helpers/mixin/instance_cache.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def __init_subclass__(
    cls,
    *,
    cache_id_attr: str | None = None,
    cache_id_func: str | None = None,
    **kwargs: Any,
) -> None:
    """Initialise subclasses with instance cache and cache ID attributes.

    Args:
        cache_id_attr (str | None, optional): The name of the attribute to use in the cache ID. Defaults to
            None. Must be provided if `cache_id_func` is None.
        cache_id_func (str | None, optional): The name of the function whose return value to use in the cache
            ID. Defaults to "__hash__" (and must be provided) if `cache_id_attr` is None.
        kwargs (Any): Additional keyword arguments to pass to the superclass.

    Raises:
        InstanceCacheSubclassError: If neither `cache_id_attr` nor `cache_id_func` are provided.
    """
    if not (cache_id_attr or cache_id_func):
        cache_id_func = "__hash__"

    if not (isinstance(cache_id_attr, str) or cache_id_attr is None):
        raise InstanceCacheSubclassError(
            f"Invalid cache ID attribute: {cache_id_attr!r}",
        )

    if not (isinstance(cache_id_func, str) or cache_id_func is None):
        raise InstanceCacheSubclassError(
            f"Invalid cache ID function: {cache_id_func!r}",
        )

    super().__init_subclass__(**kwargs)
    cls._INSTANCES = {}

    if cache_id_attr:
        cls._CACHE_ID_ATTR = cache_id_attr

    if cache_id_func:
        cls._CACHE_ID_FUNC = cache_id_func

__post_init__()

Add the instance to the cache.

Raises:

Type Description
InstanceCacheDuplicateError

If the instance already exists in the cache.
Source code in wg_utilities/helpers/mixin/instance_cache.py 
134
135
136
137
138
139
140
141
142
143
def __post_init__(self) -> None:
    """Add the instance to the cache.

    Raises:
        InstanceCacheDuplicateError: If the instance already exists in the cache.
    """
    if (instance_id := cache_id(self)) in self._INSTANCES:
        raise InstanceCacheDuplicateError(self.__class__, instance_id)

    self._INSTANCES[instance_id] = self  # type: ignore[assignment]

from_cache(cache_id_) classmethod

Get an instance from the cache by its cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
145
146
147
148
149
150
151
152
@final
@classmethod
def from_cache(cls, cache_id_: CacheIdType, /) -> Self:
    """Get an instance from the cache by its cache ID."""
    try:
        return cls._INSTANCES[cache_id_]  # type: ignore[return-value]
    except KeyError:
        raise CacheIdNotFoundError(cls, cache_id_) from None

has_cache_entry(cache_id_) classmethod

Check if the cache has an entry for the given cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
154
155
156
157
158
@final
@classmethod
def has_cache_entry(cls, cache_id_: CacheIdType, /) -> bool:
    """Check if the cache has an entry for the given cache ID."""
    return cache_id_ in cls._INSTANCES

InstanceCacheDuplicateError

Bases: InstanceCacheSubclassError

Raised when a key is already in the cache.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
21
22
23
24
25
26
27
28
29
30
class InstanceCacheDuplicateError(InstanceCacheSubclassError):
    """Raised when a key is already in the cache."""

    def __init__(self, cls: type, key: CacheIdType, /) -> None:
        self.cls = cls
        self.key = key

        super().__init__(
            f"{cls.__name__!r} instance with cache ID {key!r} already exists.",
        )

InstanceCacheIdError

Bases: InstanceCacheError

Raised when there is an error handling a cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class InstanceCacheIdError(InstanceCacheError):
    """Raised when there is an error handling a cache ID."""

    def __init__(
        self,
        cls: type,
        cache_id_: CacheIdType,
        /,
        msg: str = "Unable to process cache ID: `{cache_id}`",
    ) -> None:
        self.cls = cls
        self.cache_id = cache_id_

        cache_id_str = (
            "|".join(map(str, cache_id_))
            if isinstance(cache_id_, tuple)
            else str(cache_id_)
        )

        super().__init__(msg.format(cache_id=cache_id_str))

instance_cache

Mixin class to provide instance caching functionality.

CacheIdGenerationError

Bases: InstanceCacheIdError

Raised when there is an error generating a cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
66
67
68
69
70
71
72
73
74
class CacheIdGenerationError(InstanceCacheIdError):
    """Raised when there is an error generating a cache ID."""

    def __init__(self, cls: type, cache_id_: CacheIdType, /) -> None:
        super().__init__(
            cls,
            cache_id_,
            msg=f"Error generating cache ID for class {cls.__name__!r}: `{{cache_id}}`",
        )

CacheIdNotFoundError

Bases: InstanceCacheIdError

Raised when an ID is not found in the cache.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
55
56
57
58
59
60
61
62
63
class CacheIdNotFoundError(InstanceCacheIdError):
    """Raised when an ID is not found in the cache."""

    def __init__(self, cls: type, cache_id_: CacheIdType, /) -> None:
        super().__init__(
            cls,
            cache_id_,
            msg=f"No matching {cls.__name__!r} instance for cache ID: `{{cache_id}}`",
        )

InstanceCacheDuplicateError

Bases: InstanceCacheSubclassError

Raised when a key is already in the cache.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
21
22
23
24
25
26
27
28
29
30
class InstanceCacheDuplicateError(InstanceCacheSubclassError):
    """Raised when a key is already in the cache."""

    def __init__(self, cls: type, key: CacheIdType, /) -> None:
        self.cls = cls
        self.key = key

        super().__init__(
            f"{cls.__name__!r} instance with cache ID {key!r} already exists.",
        )

InstanceCacheError

Bases: WGUtilitiesError

Base class for all instance cache exceptions.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
13
14
class InstanceCacheError(WGUtilitiesError):
    """Base class for all instance cache exceptions."""

InstanceCacheIdError

Bases: InstanceCacheError

Raised when there is an error handling a cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class InstanceCacheIdError(InstanceCacheError):
    """Raised when there is an error handling a cache ID."""

    def __init__(
        self,
        cls: type,
        cache_id_: CacheIdType,
        /,
        msg: str = "Unable to process cache ID: `{cache_id}`",
    ) -> None:
        self.cls = cls
        self.cache_id = cache_id_

        cache_id_str = (
            "|".join(map(str, cache_id_))
            if isinstance(cache_id_, tuple)
            else str(cache_id_)
        )

        super().__init__(msg.format(cache_id=cache_id_str))

InstanceCacheMixin

Mixin class to provide instance caching functionality.

Attributes:

Name Type Description
cache_id_attr str | None

The name of the attribute to use in the cache ID. Defaults to None. Must be provided if cache_id_func is None.
cache_id_func str | None

The name of the function whose return value to use in the cache ID. Defaults to "hash". Must not be None if cache_id_attr is None.
kwargs Any

Additional keyword arguments to pass to the superclass.
Source code in wg_utilities/helpers/mixin/instance_cache.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class InstanceCacheMixin(metaclass=PostInitMeta):
    """Mixin class to provide instance caching functionality.

    Attributes:
        cache_id_attr (str | None): The name of the attribute to use in the cache ID. Defaults to None.
            Must be provided if `cache_id_func` is None.
        cache_id_func (str | None): The name of the function whose return value to use in the cache ID.
            Defaults to "__hash__". Must not be `None` if `cache_id_attr` is None.
        kwargs (Any): Additional keyword arguments to pass to the superclass.
    """

    _INSTANCES: dict[CacheIdType, Self]

    _CACHE_ID_ATTR: ClassVar[str]
    _CACHE_ID_FUNC: ClassVar[str]

    def __init_subclass__(
        cls,
        *,
        cache_id_attr: str | None = None,
        cache_id_func: str | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialise subclasses with instance cache and cache ID attributes.

        Args:
            cache_id_attr (str | None, optional): The name of the attribute to use in the cache ID. Defaults to
                None. Must be provided if `cache_id_func` is None.
            cache_id_func (str | None, optional): The name of the function whose return value to use in the cache
                ID. Defaults to "__hash__" (and must be provided) if `cache_id_attr` is None.
            kwargs (Any): Additional keyword arguments to pass to the superclass.

        Raises:
            InstanceCacheSubclassError: If neither `cache_id_attr` nor `cache_id_func` are provided.
        """
        if not (cache_id_attr or cache_id_func):
            cache_id_func = "__hash__"

        if not (isinstance(cache_id_attr, str) or cache_id_attr is None):
            raise InstanceCacheSubclassError(
                f"Invalid cache ID attribute: {cache_id_attr!r}",
            )

        if not (isinstance(cache_id_func, str) or cache_id_func is None):
            raise InstanceCacheSubclassError(
                f"Invalid cache ID function: {cache_id_func!r}",
            )

        super().__init_subclass__(**kwargs)
        cls._INSTANCES = {}

        if cache_id_attr:
            cls._CACHE_ID_ATTR = cache_id_attr

        if cache_id_func:
            cls._CACHE_ID_FUNC = cache_id_func

    def __post_init__(self) -> None:
        """Add the instance to the cache.

        Raises:
            InstanceCacheDuplicateError: If the instance already exists in the cache.
        """
        if (instance_id := cache_id(self)) in self._INSTANCES:
            raise InstanceCacheDuplicateError(self.__class__, instance_id)

        self._INSTANCES[instance_id] = self  # type: ignore[assignment]

    @final
    @classmethod
    def from_cache(cls, cache_id_: CacheIdType, /) -> Self:
        """Get an instance from the cache by its cache ID."""
        try:
            return cls._INSTANCES[cache_id_]  # type: ignore[return-value]
        except KeyError:
            raise CacheIdNotFoundError(cls, cache_id_) from None

    @final
    @classmethod
    def has_cache_entry(cls, cache_id_: CacheIdType, /) -> bool:
        """Check if the cache has an entry for the given cache ID."""
        return cache_id_ in cls._INSTANCES
__init_subclass__(*, cache_id_attr=None, cache_id_func=None, **kwargs)

Initialise subclasses with instance cache and cache ID attributes.

Parameters:

Name Type Description Default
cache_id_attr str | None

The name of the attribute to use in the cache ID. Defaults to None. Must be provided if cache_id_func is None.

 None
cache_id_func str | None

The name of the function whose return value to use in the cache ID. Defaults to "hash" (and must be provided) if cache_id_attr is None.

 None
kwargs Any

Additional keyword arguments to pass to the superclass.

 {}

Raises:

Type Description
InstanceCacheSubclassError

If neither cache_id_attr nor cache_id_func are provided.
Source code in wg_utilities/helpers/mixin/instance_cache.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def __init_subclass__(
    cls,
    *,
    cache_id_attr: str | None = None,
    cache_id_func: str | None = None,
    **kwargs: Any,
) -> None:
    """Initialise subclasses with instance cache and cache ID attributes.

    Args:
        cache_id_attr (str | None, optional): The name of the attribute to use in the cache ID. Defaults to
            None. Must be provided if `cache_id_func` is None.
        cache_id_func (str | None, optional): The name of the function whose return value to use in the cache
            ID. Defaults to "__hash__" (and must be provided) if `cache_id_attr` is None.
        kwargs (Any): Additional keyword arguments to pass to the superclass.

    Raises:
        InstanceCacheSubclassError: If neither `cache_id_attr` nor `cache_id_func` are provided.
    """
    if not (cache_id_attr or cache_id_func):
        cache_id_func = "__hash__"

    if not (isinstance(cache_id_attr, str) or cache_id_attr is None):
        raise InstanceCacheSubclassError(
            f"Invalid cache ID attribute: {cache_id_attr!r}",
        )

    if not (isinstance(cache_id_func, str) or cache_id_func is None):
        raise InstanceCacheSubclassError(
            f"Invalid cache ID function: {cache_id_func!r}",
        )

    super().__init_subclass__(**kwargs)
    cls._INSTANCES = {}

    if cache_id_attr:
        cls._CACHE_ID_ATTR = cache_id_attr

    if cache_id_func:
        cls._CACHE_ID_FUNC = cache_id_func
__post_init__()

Add the instance to the cache.

Raises:

Type Description
InstanceCacheDuplicateError

If the instance already exists in the cache.
Source code in wg_utilities/helpers/mixin/instance_cache.py 
134
135
136
137
138
139
140
141
142
143
def __post_init__(self) -> None:
    """Add the instance to the cache.

    Raises:
        InstanceCacheDuplicateError: If the instance already exists in the cache.
    """
    if (instance_id := cache_id(self)) in self._INSTANCES:
        raise InstanceCacheDuplicateError(self.__class__, instance_id)

    self._INSTANCES[instance_id] = self  # type: ignore[assignment]
from_cache(cache_id_) classmethod

Get an instance from the cache by its cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
145
146
147
148
149
150
151
152
@final
@classmethod
def from_cache(cls, cache_id_: CacheIdType, /) -> Self:
    """Get an instance from the cache by its cache ID."""
    try:
        return cls._INSTANCES[cache_id_]  # type: ignore[return-value]
    except KeyError:
        raise CacheIdNotFoundError(cls, cache_id_) from None
has_cache_entry(cache_id_) classmethod

Check if the cache has an entry for the given cache ID.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
154
155
156
157
158
@final
@classmethod
def has_cache_entry(cls, cache_id_: CacheIdType, /) -> bool:
    """Check if the cache has an entry for the given cache ID."""
    return cache_id_ in cls._INSTANCES

InstanceCacheSubclassError

Bases: InstanceCacheError

Raised when a subclass is declared incorrectly.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
17
18
class InstanceCacheSubclassError(InstanceCacheError):
    """Raised when a subclass is declared incorrectly."""

cache_id(obj)

Get the cache ID for the given object.

Source code in wg_utilities/helpers/mixin/instance_cache.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def cache_id(obj: InstanceCacheMixin, /) -> CacheIdType:
    """Get the cache ID for the given object."""
    attr_id = getattr(obj, getattr(obj, "_CACHE_ID_ATTR", "_"), None)
    func_id = getattr(obj, getattr(obj, "_CACHE_ID_FUNC", "_"), lambda: None)()

    match (attr_id, func_id):
        case (a_id, None) if a_id is not None:
            return a_id  # type: ignore[no-any-return]
        case (None, f_id) if f_id is not None:
            return f_id  # type: ignore[no-any-return]
        case (a_id, f_id) if a_id is not None and f_id is not None:
            return (a_id, f_id)
        case _:
            raise CacheIdGenerationError(obj.__class__, (attr_id, func_id))

processor

JSONProcessor

Bases: InstanceCache

Recursively process JSON objects with user-defined callbacks.

Attributes:

Name Type Description
_cbs dict

A mapping of types to a list of callback functions to be executed on the values of the given type.
identifier str

A unique identifier for the JSONProcessor instance. Defaults to the hash of the instance.
process_subclasses bool

Whether to (also) process subclasses of the target types. Defaults to True.
process_type_changes bool

Whether to re-proce_get_itemss values if their type is updated. Can lead to recursion errors if not handled correctly. Defaults to False.
process_pydantic_computed_fields bool

Whether to process fields that are computed alongside the regular model fields. Defaults to False. Only applicable if Pydantic is installed.
process_pydantic_extra_fields bool

Whether to process fields that are not explicitly defined in the Pydantic model. Only works for models with model_config.extra="allow". Defaults to False. Only applicable if Pydantic is installed.
process_pydantic_model_properties bool

Whether to process Pydantic model properties alongside the model fields. Defaults to False. Only applicable if Pydantic is installed.
ignored_loc_lookup_errors tuple

Exception types to ignore when looking up locations in the JSON object. Defaults to an empty tuple.
Source code in wg_utilities/helpers/processor/json.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
class JSONProcessor(mixin.InstanceCache, cache_id_attr="identifier"):
    """Recursively process JSON objects with user-defined callbacks.

    Attributes:
        _cbs (dict): A mapping of types to a list of callback functions to be executed on
            the values of the given type.
        identifier (str): A unique identifier for the JSONProcessor instance. Defaults to
            the hash of the instance.
        process_subclasses (bool): Whether to (also) process subclasses of the target types.
            Defaults to True.
        process_type_changes (bool): Whether to re-proce_get_itemss values if their type is updated.
            Can lead to recursion errors if not handled correctly. Defaults to False.
        process_pydantic_computed_fields (bool): Whether to process fields that are computed
            alongside the regular model fields. Defaults to False. Only applicable if Pydantic is
            installed.
        process_pydantic_extra_fields (bool): Whether to process fields that are not explicitly
            defined in the Pydantic model. Only works for models with `model_config.extra="allow"`.
            Defaults to False. Only applicable if Pydantic is installed.
        process_pydantic_model_properties (bool): Whether to process Pydantic model properties
            alongside the model fields. Defaults to False. Only applicable if Pydantic is installed.
        ignored_loc_lookup_errors (tuple): Exception types to ignore when looking up locations in
            the JSON object. Defaults to an empty tuple.
    """

    _DECORATED_CALLBACKS: ClassVar[set[Callback[..., Any]]] = set()
    __SENTINEL: Final[Sentinel] = Sentinel()

    class CallbackDefinition(NamedTuple):
        """A named tuple to hold the callback function and its associated data.

        Attributes:
            callback (Callback): The callback function to execute on the target values.
            item_filter (ItemFilter | None): An optional function to use to filter target
                values before processing them. Defaults to None.
            allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
        """

        callback: Callback[..., Any]
        item_filter: ItemFilter | None = None
        allow_callback_failures: bool = False

    CallbackMapping = dict[type[T] | type[None], list[CallbackDefinition]]

    GetterDefinition = Callable[[T, Any], Any]
    GetterMapping = dict[type[T], GetterDefinition[T]]

    IteratorFactory = Callable[[T], Iterator[Any]]
    IteratorFactoryMapping = dict[type[T], IteratorFactory[T]]

    CallbackMappingInput = dict[
        type[Any] | None,
        Callback[..., Any]
        | CallbackDefinition
        | Collection[
            Callback[..., Any]
            | tuple[Callback[..., Any],]
            | tuple[Callback[..., Any], ItemFilter]
            | tuple[Callback[..., Any], ItemFilter, bool]
            | CallbackDefinition,
        ],
    ]

    class Break(wg_exc.WGUtilitiesError):
        """Escape hatch to allow breaking out of the processing loop from within a callback."""

    def __init__(
        self,
        _cbs: CallbackMappingInput | None = None,
        /,
        *,
        identifier: str = "",
        process_subclasses: bool = True,
        process_type_changes: bool = False,
        process_pydantic_computed_fields: bool = False,
        process_pydantic_extra_fields: bool = False,
        process_pydantic_model_properties: bool = False,
        ignored_loc_lookup_errors: tuple[type[Exception], ...] = (),
    ) -> None:
        """Initialize the JSONProcessor."""
        self.callback_mapping: JSONProcessor.CallbackMapping[Any] = defaultdict(list)
        self.iterator_factory_mapping: JSONProcessor.IteratorFactoryMapping[Any] = {}
        self.getter_mapping: JSONProcessor.GetterMapping[Any] = {}

        self.config = Config(
            process_subclasses=process_subclasses,
            process_type_changes=process_type_changes,
            process_pydantic_computed_fields=process_pydantic_computed_fields,
            process_pydantic_extra_fields=process_pydantic_extra_fields,
            process_pydantic_model_properties=process_pydantic_model_properties,
            ignored_loc_lookup_errors=ignored_loc_lookup_errors,
        )

        self.identifier = identifier or hash(self)

        if _cbs:
            for target_type, cb_val in _cbs.items():
                cb_list: list[JSONProcessor.CallbackDefinition] = []
                if callable(cb_val):
                    # Single callback provided for type
                    cb_list.append(self.cb(cb_val))
                elif isinstance(cb_val, JSONProcessor.CallbackDefinition):
                    # Single CallbackDefinition named tuple provided for type
                    cb_list.append(cb_val)
                elif isinstance(cb_val, Collection):
                    for cb in cb_val:
                        if callable(cb):
                            # Single callbacks provided for type
                            cb_list.append(self.cb(cb))
                        elif isinstance(cb, tuple):
                            # Partial (or full) CallbackDefinition
                            cb_list.append(self.cb(*cb))
                        else:
                            raise InvalidCallbackError(cb, type(cb), callback=cb)
                else:
                    raise InvalidCallbackError(cb_val, type(cb_val), callback=cb_val)

                for cb_def in cb_list:
                    self.register_callback(target_type, cb_def)

        self.processable_types: tuple[type[Any], ...] = (Mapping, Sequence)
        self.unprocessable_types: tuple[type[Any], ...] = (
            (str, bytes) if self.config.process_subclasses else ()
        )

    @staticmethod
    def cb(
        callback: Callback[..., Any],
        item_filter: ItemFilter | None = None,
        allow_callback_failures: bool = False,  # noqa: FBT001,FBT002
        *,
        allow_mutation: bool = True,
    ) -> CallbackDefinition:
        """Create a CallbackDefinition for the given callback.

        Args:
            callback (Callback): The callback function to execute on the target values. Can take None, any, or all
                of the following arguments: `_value_`, `_loc_`, `_obj_type_`, `_depth_`, and any additional keyword
                arguments, which will be passed in from the `JSONProcessor.process` method.
            item_filter (ItemFilter | None): An optional function to use to filter target values before processing
                them. Defaults to None. Function signature is as follows:
                ```python
                def item_filter(item: Any, /, *, loc: K | int | str) -> bool: ...
                ```
            allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
            allow_mutation (bool): Whether the callback is allowed to mutate the input object. Defaults to True.
        """

        if callback.__name__ == "<lambda>":
            callback = JSONProcessor.callback(allow_mutation=allow_mutation)(callback)

        if item_filter and not callable(item_filter):
            raise InvalidItemFilterError(item_filter, callback=callback)

        if not isinstance(allow_callback_failures, bool):
            raise InvalidCallbackError(
                allow_callback_failures,
                type(allow_callback_failures),
                callback=callback,
            )

        return JSONProcessor.CallbackDefinition(
            callback=callback,
            item_filter=item_filter,
            allow_callback_failures=allow_callback_failures,
        )

    def _get_callbacks(
        self,
        typ: type[Any],
    ) -> Generator[CallbackDefinition, None, None]:
        if self.config.process_subclasses:
            for cb_typ in self.callback_mapping:
                if issubclass(typ, cb_typ):
                    yield from self.callback_mapping[cb_typ]
        elif callback_list := self.callback_mapping.get(typ):
            yield from callback_list

    @overload
    def _get_getter_or_iterator_factory(
        self,
        typ: type[T],
        mapping: JSONProcessor.IteratorFactoryMapping[Any],
    ) -> JSONProcessor.IteratorFactory[T] | None: ...

    @overload
    def _get_getter_or_iterator_factory(
        self,
        typ: type[T],
        mapping: JSONProcessor.GetterMapping[Any],
    ) -> JSONProcessor.GetterDefinition[T] | None: ...

    def _get_getter_or_iterator_factory(
        self,
        typ: type[T],
        mapping: JSONProcessor.IteratorFactoryMapping[Any]
        | JSONProcessor.GetterMapping[Any],
    ) -> JSONProcessor.IteratorFactory[T] | JSONProcessor.GetterDefinition[T] | None:
        """Get the getter or iterator for the given type.

        If `config.process_subclasses` is true, the getter or iterator for the first matching
        subclass will be returned. Otherwise, the getter or iterator for the exact type will be
        returned.
        """

        if (exact_match := mapping.get(typ)) or not self.config.process_subclasses:
            return exact_match

        for key, value in mapping.items():
            if issubclass(typ, key):
                return value

        return None

    @overload
    def _get_item(self, obj: BaseModel, loc: str) -> object: ...

    @overload
    def _get_item(self, obj: Mapping[K, object], loc: K) -> object: ...

    @overload
    def _get_item(self, obj: Sequence[object], loc: int) -> object: ...

    @overload
    def _get_item(self, obj: G, loc: object) -> object: ...

    def _get_item(
        self,
        obj: BaseModel | Mapping[K, object] | Sequence[object] | G,
        loc: str | K | int | object,
    ) -> Any | Sentinel:
        with suppress(*self.config.ignored_loc_lookup_errors):
            try:
                if custom_getter := self._get_getter_or_iterator_factory(
                    type(obj), self.getter_mapping
                ):
                    return custom_getter(obj, loc)

                try:
                    return obj[loc]  # type: ignore[index]
                except TypeError as exc:
                    if not isinstance(loc, str) or (
                        not str(exc).endswith("is not subscriptable")
                        and str(exc) != "list indices must be integers or slices, not str"
                    ):
                        raise

                try:
                    return getattr(obj, loc)
                except AttributeError:
                    raise LocNotFoundError(loc, obj) from None

            except LookupError as exc:
                raise LocNotFoundError(loc, obj) from exc

        return JSONProcessor.__SENTINEL

    @staticmethod
    def _set_item(
        obj: Mapping[K, V] | Sequence[object] | BaseModel,
        loc: str | K | int,
        val: V,
    ) -> None:
        with suppress(TypeError):
            obj[loc] = val  # type: ignore[index]
            return

        if not isinstance(loc, str):
            return

        with suppress(AttributeError):
            setattr(obj, loc, val)

        return

    @overload
    def _iterate(self, obj: Mapping[K, Any]) -> Iterator[K]: ...

    @overload
    def _iterate(self, obj: Sequence[Any]) -> Iterator[int]: ...

    @overload
    def _iterate(self, obj: BaseModel) -> Iterator[str]: ...

    def _iterate(
        self,
        obj: Mapping[K, Any] | Sequence[Any] | BaseModel,
    ) -> Iterator[K] | Iterator[int] | Iterator[str] | Sentinel:
        if custom_iter := self._get_getter_or_iterator_factory(
            type(obj), self.iterator_factory_mapping
        ):
            return custom_iter(obj)

        if isinstance(obj, Sequence):
            return iter(range(len(obj)))

        if isinstance(obj, BaseModel):
            iterables: list[Iterable[str]] = [obj.model_fields.keys()]

            if self.config.process_pydantic_model_properties:
                iterables.append(
                    sorted(
                        {
                            name
                            for name, prop in inspect.getmembers(obj.__class__)
                            if isinstance(prop, property)
                            and name not in BASE_MODEL_PROPERTIES
                        },
                    ),
                )

            if self.config.process_pydantic_computed_fields:
                iterables.append(obj.model_computed_fields.keys())

            if self.config.process_pydantic_extra_fields and obj.model_extra:
                iterables.append(obj.model_extra.keys())

            return iter(chain(*iterables))

        try:
            return iter(obj)
        except TypeError as exc:
            if not str(exc).endswith("is not iterable"):
                raise

        return self.__SENTINEL

    @overload
    def _process_item(
        self,
        *,
        obj: BaseModel,
        loc: str,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_item(
        self,
        *,
        obj: Mapping[K, object],
        loc: K,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_item(
        self,
        *,
        obj: Sequence[object],
        loc: int,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None: ...

    def _process_item(
        self,
        *,
        obj: BaseModel | Mapping[K, object] | Sequence[object],
        loc: str | K | int,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None:
        try:
            out = cb(
                _value_=self._get_item(obj, loc),
                _loc_=loc,
                _obj_type_=type(obj),
                _depth_=depth,
                **kwargs,
            )
        except TypeError as exc:
            arg_error: type[InvalidCallbackArgumentsError]
            for arg_error in InvalidCallbackArgumentsError.subclasses():  # type: ignore[assignment]
                if match := arg_error.PATTERN.search(str(exc)):
                    raise arg_error(*match.groups(), callback=cb) from None

            raise

        if out is not self.__SENTINEL:
            self._set_item(obj, loc, out)

            if self.config.process_type_changes and type(out) != orig_item_type:
                self._process_loc(
                    obj=obj,  # type: ignore[arg-type]
                    loc=loc,  # type: ignore[arg-type]
                    depth=depth,
                    kwargs=kwargs,
                )

    @overload
    def _process_loc(
        self,
        *,
        obj: BaseModel,
        loc: str,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_loc(
        self,
        *,
        obj: Sequence[object],
        loc: int,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_loc(
        self,
        *,
        obj: Mapping[K, object],
        loc: K,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None: ...

    def _process_loc(
        self,
        *,
        obj: BaseModel | Mapping[K, object] | Sequence[object],
        loc: str | K | int,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None:
        try:
            item = self._get_item(obj, loc)
        except LocNotFoundError:
            return

        for cb, item_filter, allow_failures in self._get_callbacks(
            orig_item_type := type(item),
        ):
            if not item_filter or bool(item_filter(item, loc=loc)):
                try:
                    self._process_item(
                        obj=obj,  # type: ignore[arg-type]
                        loc=loc,  # type: ignore[arg-type]
                        cb=cb,
                        depth=depth,
                        orig_item_type=orig_item_type,
                        kwargs=kwargs,
                    )
                except (self.Break, InvalidCallbackArgumentsError):
                    raise
                except Exception:
                    if not allow_failures:
                        raise

    def process(
        self,
        obj: Mapping[K, object] | Sequence[object],
        /,
        __depth: int = 0,
        __processed_models: set[BaseModel] | None = None,
        **kwargs: Any,
    ) -> None:
        """Recursively process a JSON object with the registered callbacks.

        Args:
            obj: The JSON object to process.
            kwargs: Any additional keyword arguments to pass to the callback(s).
        """

        for loc in self._iterate(obj):
            try:
                self._process_loc(
                    obj=obj,
                    loc=loc,
                    depth=__depth,
                    kwargs=kwargs,
                )
            except self.Break:
                break

            item = self._get_item(obj, loc)

            if isinstance(item, self.processable_types) and not isinstance(
                item,
                self.unprocessable_types,
            ):
                self.process(
                    item,
                    _JSONProcessor__depth=__depth + 1,
                    _JSONProcessor__processed_models=__processed_models,
                    **kwargs,
                )
            elif (
                PYDANTIC_INSTALLED
                and isinstance(item, BaseModel)
                and item not in (__processed_models or set())
            ):
                self.process_model(
                    item,
                    _JSONProcessor__depth=__depth + 1,
                    _JSONProcessor__processed_models=__processed_models,
                    **kwargs,
                )

    def process_anything(self, obj: Any, **kwargs: Any) -> None:  # pragma: no cover
        """Process anything that can be processed."""
        self.process(obj, **kwargs)

    if PYDANTIC_INSTALLED:

        def process_model(
            self,
            model: BaseModel,
            /,
            __depth: int = 0,
            __processed_models: set[BaseModel] | None = None,
            **kwargs: Any,
        ) -> None:
            """Recursively process a Pydantic model with the registered callbacks.

            Args:
                model: The Pydantic model to process.
                kwargs: Any additional keyword arguments to pass to the callback(s).
            """
            if __processed_models is None:
                __processed_models = {model}
            else:
                __processed_models.add(model)

            for loc in self._iterate(model):
                try:
                    self._process_loc(obj=model, loc=loc, depth=__depth, kwargs=kwargs)
                except self.Break:
                    break

                try:
                    item = getattr(model, loc)
                except self.config.ignored_loc_lookup_errors:
                    continue

                if isinstance(item, BaseModel) and item not in __processed_models:
                    self.process_model(
                        item,
                        _JSONProcessor__depth=__depth + 1,
                        _JSONProcessor__processed_models=__processed_models,
                        **kwargs,
                    )
                elif isinstance(item, Mapping | Sequence) and not isinstance(
                    item,
                    (str, bytes),
                ):
                    self.process(
                        item,
                        _JSONProcessor__depth=__depth + 1,
                        _JSONProcessor__processed_models=__processed_models,
                        **kwargs,
                    )

    def register_callback(
        self,
        target_type: type[C] | None,
        callback_def: CallbackDefinition,
    ) -> None:
        """Register a new callback for use when processing any JSON objects.

        Args:
            target_type (type): The type of the values to be processed.
            callback_def (CallbackDefinition): The callback definition to register.
        """
        decorated = (
            callback_def.callback.__func__
            if inspect.ismethod(callback_def.callback)
            else callback_def.callback
        )

        if decorated not in JSONProcessor._DECORATED_CALLBACKS:
            raise CallbackNotDecoratedError(decorated)

        self.callback_mapping[target_type or type(None)].append(callback_def)

    def register_custom_getter(
        self,
        target_type: type[G],
        getter: GetterDefinition[G],
        *,
        add_to_processable_type_list: bool = True,
    ) -> None:
        """Register a custom getter for use when processing any JSON objects.

        Args:
            target_type (type): The type of the values to be processed.
            getter (Callable): The custom getter to register.
            add_to_processable_type_list (bool): Whether to add the target type to the list of
                processable types. Defaults to True.
        """
        self.getter_mapping[target_type] = getter

        if add_to_processable_type_list and target_type not in self.processable_types:
            self.processable_types = (*self.processable_types, target_type)

    def register_custom_iterator(
        self,
        target_type: type[I],
        iterator: IteratorFactory[I],
        *,
        add_to_processable_type_list: bool = True,
    ) -> None:
        """Register a custom iterator for use when processing any JSON objects.

        Args:
            target_type (type): The type of the values to be processed.
            iterator (Callable): The custom iterator to register.
            add_to_processable_type_list (bool): Whether to add the target type to the list of
                processable types. Defaults to True.
        """
        self.iterator_factory_mapping[target_type] = iterator

        if add_to_processable_type_list and target_type not in self.processable_types:
            self.processable_types = (*self.processable_types, target_type)

    @overload
    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: Literal[True] = True,
    ) -> Callable[[Callable[P, R]], Callback[P, R]]: ...

    @overload
    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: Literal[False],
    ) -> Callable[[Callable[P, R]], Callback[P, Sentinel]]: ...

    @overload
    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: bool = ...,
    ) -> Callable[[Callable[P, R]], Callback[P, R | Sentinel]]: ...

    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: bool = True,
    ) -> Callable[[Callable[P, R]], Callback[P, R | Sentinel]]:
        """Decorator to mark a function as a callback for use with the JSONProcessor.

        Warning:
            `allow_mutation` only blocks the return value from being used to update the input
            object. It does not prevent the callback from mutating the input object (or other
            objects passed in as arguments) in place.

        Args:
            allow_mutation (bool): Whether the callback is allowed to mutate the input object.
                Defaults to True.
        """

        def _decorator(func: Callable[P, R]) -> Callback[P, R | Sentinel]:
            """Decorator to mark a function as a callback for use with the JSONProcessor."""

            if isinstance(func, classmethod):
                raise InvalidCallbackError(
                    "@JSONProcessor.callback must be used _after_ @classmethod",
                    callback=func,
                )

            arg_names, kwarg_names = [], []

            for name, param in inspect.signature(func).parameters.items():
                if param.kind == param.POSITIONAL_ONLY:
                    arg_names.append(name)
                else:
                    kwarg_names.append(name)

            def filter_kwargs(
                kwargs: dict[str, Any],
                /,
            ) -> tuple[list[Any], dict[str, Any]]:
                a = []
                for an in arg_names:
                    with suppress(KeyError):
                        a.append(kwargs[an])

                kw = {}
                for kn in kwarg_names:
                    with suppress(KeyError):
                        kw[kn] = kwargs[kn]

                return a, kw

            # This is the same `cb` called in the `JSONProcessor._process_loc` method - no positional
            # arguments are explicitly passed in (and would be rejected by `JSONProcessor.process` anyway),
            # unless the callback is a bound method. In this case, the `cls`/`self` argument is passed in
            # as the first positional argument, hence the need for `*bound_args` below

            if allow_mutation:

                @wraps(func)
                def cb(*bound_args: P.args, **process_kwargs: P.kwargs) -> R:
                    args, kwargs = filter_kwargs(process_kwargs)
                    return func(*bound_args, *args, **kwargs)

            else:

                @wraps(func)
                def cb(
                    *bound_args: P.args,
                    **process_kwargs: P.kwargs,
                ) -> Sentinel:
                    args, kwargs = filter_kwargs(process_kwargs)
                    func(*bound_args, *args, **kwargs)
                    return JSONProcessor.__SENTINEL

            cls._DECORATED_CALLBACKS.add(cb)

            return cb

        return _decorator

Break

Bases: WGUtilitiesError

Escape hatch to allow breaking out of the processing loop from within a callback.

Source code in wg_utilities/helpers/processor/json.py 
237
238
class Break(wg_exc.WGUtilitiesError):
    """Escape hatch to allow breaking out of the processing loop from within a callback."""

CallbackDefinition

Bases: NamedTuple

A named tuple to hold the callback function and its associated data.

Attributes:

Name Type Description
callback Callback

The callback function to execute on the target values.
item_filter ItemFilter | None

An optional function to use to filter target values before processing them. Defaults to None.
allow_callback_failures bool

Whether to allow callback failures. Defaults to False.
Source code in wg_utilities/helpers/processor/json.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
class CallbackDefinition(NamedTuple):
    """A named tuple to hold the callback function and its associated data.

    Attributes:
        callback (Callback): The callback function to execute on the target values.
        item_filter (ItemFilter | None): An optional function to use to filter target
            values before processing them. Defaults to None.
        allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
    """

    callback: Callback[..., Any]
    item_filter: ItemFilter | None = None
    allow_callback_failures: bool = False

__init__(_cbs=None, /, *, identifier='', process_subclasses=True, process_type_changes=False, process_pydantic_computed_fields=False, process_pydantic_extra_fields=False, process_pydantic_model_properties=False, ignored_loc_lookup_errors=())

Initialize the JSONProcessor.

Source code in wg_utilities/helpers/processor/json.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def __init__(
    self,
    _cbs: CallbackMappingInput | None = None,
    /,
    *,
    identifier: str = "",
    process_subclasses: bool = True,
    process_type_changes: bool = False,
    process_pydantic_computed_fields: bool = False,
    process_pydantic_extra_fields: bool = False,
    process_pydantic_model_properties: bool = False,
    ignored_loc_lookup_errors: tuple[type[Exception], ...] = (),
) -> None:
    """Initialize the JSONProcessor."""
    self.callback_mapping: JSONProcessor.CallbackMapping[Any] = defaultdict(list)
    self.iterator_factory_mapping: JSONProcessor.IteratorFactoryMapping[Any] = {}
    self.getter_mapping: JSONProcessor.GetterMapping[Any] = {}

    self.config = Config(
        process_subclasses=process_subclasses,
        process_type_changes=process_type_changes,
        process_pydantic_computed_fields=process_pydantic_computed_fields,
        process_pydantic_extra_fields=process_pydantic_extra_fields,
        process_pydantic_model_properties=process_pydantic_model_properties,
        ignored_loc_lookup_errors=ignored_loc_lookup_errors,
    )

    self.identifier = identifier or hash(self)

    if _cbs:
        for target_type, cb_val in _cbs.items():
            cb_list: list[JSONProcessor.CallbackDefinition] = []
            if callable(cb_val):
                # Single callback provided for type
                cb_list.append(self.cb(cb_val))
            elif isinstance(cb_val, JSONProcessor.CallbackDefinition):
                # Single CallbackDefinition named tuple provided for type
                cb_list.append(cb_val)
            elif isinstance(cb_val, Collection):
                for cb in cb_val:
                    if callable(cb):
                        # Single callbacks provided for type
                        cb_list.append(self.cb(cb))
                    elif isinstance(cb, tuple):
                        # Partial (or full) CallbackDefinition
                        cb_list.append(self.cb(*cb))
                    else:
                        raise InvalidCallbackError(cb, type(cb), callback=cb)
            else:
                raise InvalidCallbackError(cb_val, type(cb_val), callback=cb_val)

            for cb_def in cb_list:
                self.register_callback(target_type, cb_def)

    self.processable_types: tuple[type[Any], ...] = (Mapping, Sequence)
    self.unprocessable_types: tuple[type[Any], ...] = (
        (str, bytes) if self.config.process_subclasses else ()
    )

callback(*, allow_mutation=True) classmethod

Decorator to mark a function as a callback for use with the JSONProcessor.

Warning

allow_mutation only blocks the return value from being used to update the input object. It does not prevent the callback from mutating the input object (or other objects passed in as arguments) in place.

Parameters:

Name Type Description Default
allow_mutation bool

Whether the callback is allowed to mutate the input object. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
@classmethod
def callback(
    cls,
    *,
    allow_mutation: bool = True,
) -> Callable[[Callable[P, R]], Callback[P, R | Sentinel]]:
    """Decorator to mark a function as a callback for use with the JSONProcessor.

    Warning:
        `allow_mutation` only blocks the return value from being used to update the input
        object. It does not prevent the callback from mutating the input object (or other
        objects passed in as arguments) in place.

    Args:
        allow_mutation (bool): Whether the callback is allowed to mutate the input object.
            Defaults to True.
    """

    def _decorator(func: Callable[P, R]) -> Callback[P, R | Sentinel]:
        """Decorator to mark a function as a callback for use with the JSONProcessor."""

        if isinstance(func, classmethod):
            raise InvalidCallbackError(
                "@JSONProcessor.callback must be used _after_ @classmethod",
                callback=func,
            )

        arg_names, kwarg_names = [], []

        for name, param in inspect.signature(func).parameters.items():
            if param.kind == param.POSITIONAL_ONLY:
                arg_names.append(name)
            else:
                kwarg_names.append(name)

        def filter_kwargs(
            kwargs: dict[str, Any],
            /,
        ) -> tuple[list[Any], dict[str, Any]]:
            a = []
            for an in arg_names:
                with suppress(KeyError):
                    a.append(kwargs[an])

            kw = {}
            for kn in kwarg_names:
                with suppress(KeyError):
                    kw[kn] = kwargs[kn]

            return a, kw

        # This is the same `cb` called in the `JSONProcessor._process_loc` method - no positional
        # arguments are explicitly passed in (and would be rejected by `JSONProcessor.process` anyway),
        # unless the callback is a bound method. In this case, the `cls`/`self` argument is passed in
        # as the first positional argument, hence the need for `*bound_args` below

        if allow_mutation:

            @wraps(func)
            def cb(*bound_args: P.args, **process_kwargs: P.kwargs) -> R:
                args, kwargs = filter_kwargs(process_kwargs)
                return func(*bound_args, *args, **kwargs)

        else:

            @wraps(func)
            def cb(
                *bound_args: P.args,
                **process_kwargs: P.kwargs,
            ) -> Sentinel:
                args, kwargs = filter_kwargs(process_kwargs)
                func(*bound_args, *args, **kwargs)
                return JSONProcessor.__SENTINEL

        cls._DECORATED_CALLBACKS.add(cb)

        return cb

    return _decorator

cb(callback, item_filter=None, allow_callback_failures=False, *, allow_mutation=True) staticmethod

Create a CallbackDefinition for the given callback.

Parameters:

Name Type Description Default
callback Callback

The callback function to execute on the target values. Can take None, any, or all of the following arguments: _value_, _loc_, _obj_type_, _depth_, and any additional keyword arguments, which will be passed in from the JSONProcessor.process method.

 required
item_filter ItemFilter | None

An optional function to use to filter target values before processing them. Defaults to None. Function signature is as follows:

def item_filter(item: Any, /, *, loc: K | int | str) -> bool: ...
None
allow_callback_failures bool

Whether to allow callback failures. Defaults to False.

 False
allow_mutation bool

Whether the callback is allowed to mutate the input object. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
@staticmethod
def cb(
    callback: Callback[..., Any],
    item_filter: ItemFilter | None = None,
    allow_callback_failures: bool = False,  # noqa: FBT001,FBT002
    *,
    allow_mutation: bool = True,
) -> CallbackDefinition:
    """Create a CallbackDefinition for the given callback.

    Args:
        callback (Callback): The callback function to execute on the target values. Can take None, any, or all
            of the following arguments: `_value_`, `_loc_`, `_obj_type_`, `_depth_`, and any additional keyword
            arguments, which will be passed in from the `JSONProcessor.process` method.
        item_filter (ItemFilter | None): An optional function to use to filter target values before processing
            them. Defaults to None. Function signature is as follows:
            ```python
            def item_filter(item: Any, /, *, loc: K | int | str) -> bool: ...
            ```
        allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
        allow_mutation (bool): Whether the callback is allowed to mutate the input object. Defaults to True.
    """

    if callback.__name__ == "<lambda>":
        callback = JSONProcessor.callback(allow_mutation=allow_mutation)(callback)

    if item_filter and not callable(item_filter):
        raise InvalidItemFilterError(item_filter, callback=callback)

    if not isinstance(allow_callback_failures, bool):
        raise InvalidCallbackError(
            allow_callback_failures,
            type(allow_callback_failures),
            callback=callback,
        )

    return JSONProcessor.CallbackDefinition(
        callback=callback,
        item_filter=item_filter,
        allow_callback_failures=allow_callback_failures,
    )

process(obj, /, __depth=0, __processed_models=None, **kwargs)

Recursively process a JSON object with the registered callbacks.

Parameters:

Name Type Description Default
obj Mapping[K, object] | Sequence[object]

The JSON object to process.

 required
kwargs Any

Any additional keyword arguments to pass to the callback(s).

 {}
Source code in wg_utilities/helpers/processor/json.py 
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
def process(
    self,
    obj: Mapping[K, object] | Sequence[object],
    /,
    __depth: int = 0,
    __processed_models: set[BaseModel] | None = None,
    **kwargs: Any,
) -> None:
    """Recursively process a JSON object with the registered callbacks.

    Args:
        obj: The JSON object to process.
        kwargs: Any additional keyword arguments to pass to the callback(s).
    """

    for loc in self._iterate(obj):
        try:
            self._process_loc(
                obj=obj,
                loc=loc,
                depth=__depth,
                kwargs=kwargs,
            )
        except self.Break:
            break

        item = self._get_item(obj, loc)

        if isinstance(item, self.processable_types) and not isinstance(
            item,
            self.unprocessable_types,
        ):
            self.process(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )
        elif (
            PYDANTIC_INSTALLED
            and isinstance(item, BaseModel)
            and item not in (__processed_models or set())
        ):
            self.process_model(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )

process_anything(obj, **kwargs)

Process anything that can be processed.

Source code in wg_utilities/helpers/processor/json.py 
686
687
688
def process_anything(self, obj: Any, **kwargs: Any) -> None:  # pragma: no cover
    """Process anything that can be processed."""
    self.process(obj, **kwargs)

process_model(model, /, __depth=0, __processed_models=None, **kwargs)

Recursively process a Pydantic model with the registered callbacks.

Parameters:

Name Type Description Default
model BaseModel

The Pydantic model to process.

 required
kwargs Any

Any additional keyword arguments to pass to the callback(s).

 {}
Source code in wg_utilities/helpers/processor/json.py 
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
def process_model(
    self,
    model: BaseModel,
    /,
    __depth: int = 0,
    __processed_models: set[BaseModel] | None = None,
    **kwargs: Any,
) -> None:
    """Recursively process a Pydantic model with the registered callbacks.

    Args:
        model: The Pydantic model to process.
        kwargs: Any additional keyword arguments to pass to the callback(s).
    """
    if __processed_models is None:
        __processed_models = {model}
    else:
        __processed_models.add(model)

    for loc in self._iterate(model):
        try:
            self._process_loc(obj=model, loc=loc, depth=__depth, kwargs=kwargs)
        except self.Break:
            break

        try:
            item = getattr(model, loc)
        except self.config.ignored_loc_lookup_errors:
            continue

        if isinstance(item, BaseModel) and item not in __processed_models:
            self.process_model(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )
        elif isinstance(item, Mapping | Sequence) and not isinstance(
            item,
            (str, bytes),
        ):
            self.process(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )

register_callback(target_type, callback_def)

Register a new callback for use when processing any JSON objects.

Parameters:

Name Type Description Default
target_type type

The type of the values to be processed.

 required
callback_def CallbackDefinition

The callback definition to register.

 required
Source code in wg_utilities/helpers/processor/json.py 
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
def register_callback(
    self,
    target_type: type[C] | None,
    callback_def: CallbackDefinition,
) -> None:
    """Register a new callback for use when processing any JSON objects.

    Args:
        target_type (type): The type of the values to be processed.
        callback_def (CallbackDefinition): The callback definition to register.
    """
    decorated = (
        callback_def.callback.__func__
        if inspect.ismethod(callback_def.callback)
        else callback_def.callback
    )

    if decorated not in JSONProcessor._DECORATED_CALLBACKS:
        raise CallbackNotDecoratedError(decorated)

    self.callback_mapping[target_type or type(None)].append(callback_def)

register_custom_getter(target_type, getter, *, add_to_processable_type_list=True)

Register a custom getter for use when processing any JSON objects.

Parameters:

Name Type Description Default
target_type type

The type of the values to be processed.

 required
getter Callable

The custom getter to register.

 required
add_to_processable_type_list bool

Whether to add the target type to the list of processable types. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
def register_custom_getter(
    self,
    target_type: type[G],
    getter: GetterDefinition[G],
    *,
    add_to_processable_type_list: bool = True,
) -> None:
    """Register a custom getter for use when processing any JSON objects.

    Args:
        target_type (type): The type of the values to be processed.
        getter (Callable): The custom getter to register.
        add_to_processable_type_list (bool): Whether to add the target type to the list of
            processable types. Defaults to True.
    """
    self.getter_mapping[target_type] = getter

    if add_to_processable_type_list and target_type not in self.processable_types:
        self.processable_types = (*self.processable_types, target_type)

register_custom_iterator(target_type, iterator, *, add_to_processable_type_list=True)

Register a custom iterator for use when processing any JSON objects.

Parameters:

Name Type Description Default
target_type type

The type of the values to be processed.

 required
iterator Callable

The custom iterator to register.

 required
add_to_processable_type_list bool

Whether to add the target type to the list of processable types. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
def register_custom_iterator(
    self,
    target_type: type[I],
    iterator: IteratorFactory[I],
    *,
    add_to_processable_type_list: bool = True,
) -> None:
    """Register a custom iterator for use when processing any JSON objects.

    Args:
        target_type (type): The type of the values to be processed.
        iterator (Callable): The custom iterator to register.
        add_to_processable_type_list (bool): Whether to add the target type to the list of
            processable types. Defaults to True.
    """
    self.iterator_factory_mapping[target_type] = iterator

    if add_to_processable_type_list and target_type not in self.processable_types:
        self.processable_types = (*self.processable_types, target_type)

json

Useful functions for working with JSON/dictionaries.

BaseModel

Dummy class for when pydantic is not installed.

As long as isinstance returns False, it's all good.

Source code in wg_utilities/helpers/processor/json.py 
39
40
41
42
43
class BaseModel:  # type: ignore[no-redef]
    """Dummy class for when pydantic is not installed.

    As long as `isinstance` returns False, it's all good.
    """

CallbackNotDecoratedError

Bases: InvalidCallbackError

Raised when a callback is not decorated with @JSONProcessor.callback.

Source code in wg_utilities/helpers/processor/json.py 
114
115
116
117
118
119
120
121
122
class CallbackNotDecoratedError(InvalidCallbackError):
    """Raised when a callback is not decorated with `@JSONProcessor.callback`."""

    def __init__(self, callback: Callback[..., Any], /) -> None:
        super().__init__(
            f"Callback `{callback.__module__}.{callback.__name__}` must be decorated "
            "with `@JSONProcessor.callback`",
            callback=callback,
        )

Config dataclass

Configuration for the JSONProcessor.

Source code in wg_utilities/helpers/processor/json.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@dataclass
class Config:
    """Configuration for the JSONProcessor."""

    process_subclasses: bool
    """Whether to (also) process subclasses of the target types."""

    process_type_changes: bool
    """Whether to re-process values if their type is updated.

    Can lead to recursion errors if not handled correctly.
    """

    process_pydantic_computed_fields: bool
    """Whether to process fields that are computed alongside the regular model fields."""

    process_pydantic_extra_fields: bool
    """Whether to process fields that are not explicitly defined in the Pydantic model.

    Only works for models with `model_config.extra="allow"`
    """

    process_pydantic_model_properties: bool
    """Whether to process Pydantic model properties alongside the model fields."""

    ignored_loc_lookup_errors: tuple[type[BaseException], ...]
    """Exception types to ignore when looking up locations in the JSON object."""
ignored_loc_lookup_errors: tuple[type[BaseException], ...] instance-attribute

Exception types to ignore when looking up locations in the JSON object.

process_pydantic_computed_fields: bool instance-attribute

Whether to process fields that are computed alongside the regular model fields.

process_pydantic_extra_fields: bool instance-attribute

Whether to process fields that are not explicitly defined in the Pydantic model.

Only works for models with model_config.extra="allow"

process_pydantic_model_properties: bool instance-attribute

Whether to process Pydantic model properties alongside the model fields.

process_subclasses: bool instance-attribute

Whether to (also) process subclasses of the target types.

process_type_changes: bool instance-attribute

Whether to re-process values if their type is updated.

Can lead to recursion errors if not handled correctly.

InvalidCallbackArgumentsError

Bases: InvalidCallbackError

Raised when a callback is called with extra and/or missing keyword or positional arguments.

Source code in wg_utilities/helpers/processor/json.py 
87
88
89
90
91
92
93
94
95
96
97
class InvalidCallbackArgumentsError(InvalidCallbackError):
    """Raised when a callback is called with extra and/or missing keyword or positional arguments."""

    ARG_TYPE: ClassVar[str]
    PATTERN: ClassVar[re.Pattern[str]]

    def __init__(self, *arg_names: str, callback: Callback[..., Any]) -> None:
        super().__init__(
            f"Missing required {self.ARG_TYPE} argument(s): `{'`, `'.join(arg_names)}`.",
            callback=callback,
        )

InvalidCallbackError

Bases: BadDefinitionError

Raised for invalid callback handling.

Source code in wg_utilities/helpers/processor/json.py 
78
79
80
81
82
83
84
class InvalidCallbackError(wg_exc.BadDefinitionError):
    """Raised for invalid callback handling."""

    def __init__(self, *args: object, callback: Callback[..., Any]) -> None:
        super().__init__(*args)

        self.callback = callback

InvalidItemFilterError

Bases: InvalidCallbackError

Raised when an item filter is not callable.

Source code in wg_utilities/helpers/processor/json.py 
125
126
127
128
129
130
131
132
133
class InvalidItemFilterError(InvalidCallbackError):
    """Raised when an item filter is not callable."""

    def __init__(self, item_filter: Any, /, *, callback: Callback[..., Any]) -> None:
        super().__init__(
            f"Item filter `{item_filter!r}` is not callable ({type(item_filter)}).",
            callback=callback,
        )
        self.item_filter = item_filter

ItemFilter

Bases: Protocol

Function to filter items before processing them.

Source code in wg_utilities/helpers/processor/json.py 
136
137
138
139
140
class ItemFilter(Protocol):
    """Function to filter items before processing them."""

    def __call__(self, item: Any, /, *, loc: K | int | str) -> bool:
        """The function to be called on each value in the JSON object."""
__call__(item, /, *, loc)

The function to be called on each value in the JSON object.

Source code in wg_utilities/helpers/processor/json.py 
139
140
def __call__(self, item: Any, /, *, loc: K | int | str) -> bool:
    """The function to be called on each value in the JSON object."""

JSONProcessor

Bases: InstanceCache

Recursively process JSON objects with user-defined callbacks.

Attributes:

Name Type Description
_cbs dict

A mapping of types to a list of callback functions to be executed on the values of the given type.
identifier str

A unique identifier for the JSONProcessor instance. Defaults to the hash of the instance.
process_subclasses bool

Whether to (also) process subclasses of the target types. Defaults to True.
process_type_changes bool

Whether to re-proce_get_itemss values if their type is updated. Can lead to recursion errors if not handled correctly. Defaults to False.
process_pydantic_computed_fields bool

Whether to process fields that are computed alongside the regular model fields. Defaults to False. Only applicable if Pydantic is installed.
process_pydantic_extra_fields bool

Whether to process fields that are not explicitly defined in the Pydantic model. Only works for models with model_config.extra="allow". Defaults to False. Only applicable if Pydantic is installed.
process_pydantic_model_properties bool

Whether to process Pydantic model properties alongside the model fields. Defaults to False. Only applicable if Pydantic is installed.
ignored_loc_lookup_errors tuple

Exception types to ignore when looking up locations in the JSON object. Defaults to an empty tuple.
Source code in wg_utilities/helpers/processor/json.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
class JSONProcessor(mixin.InstanceCache, cache_id_attr="identifier"):
    """Recursively process JSON objects with user-defined callbacks.

    Attributes:
        _cbs (dict): A mapping of types to a list of callback functions to be executed on
            the values of the given type.
        identifier (str): A unique identifier for the JSONProcessor instance. Defaults to
            the hash of the instance.
        process_subclasses (bool): Whether to (also) process subclasses of the target types.
            Defaults to True.
        process_type_changes (bool): Whether to re-proce_get_itemss values if their type is updated.
            Can lead to recursion errors if not handled correctly. Defaults to False.
        process_pydantic_computed_fields (bool): Whether to process fields that are computed
            alongside the regular model fields. Defaults to False. Only applicable if Pydantic is
            installed.
        process_pydantic_extra_fields (bool): Whether to process fields that are not explicitly
            defined in the Pydantic model. Only works for models with `model_config.extra="allow"`.
            Defaults to False. Only applicable if Pydantic is installed.
        process_pydantic_model_properties (bool): Whether to process Pydantic model properties
            alongside the model fields. Defaults to False. Only applicable if Pydantic is installed.
        ignored_loc_lookup_errors (tuple): Exception types to ignore when looking up locations in
            the JSON object. Defaults to an empty tuple.
    """

    _DECORATED_CALLBACKS: ClassVar[set[Callback[..., Any]]] = set()
    __SENTINEL: Final[Sentinel] = Sentinel()

    class CallbackDefinition(NamedTuple):
        """A named tuple to hold the callback function and its associated data.

        Attributes:
            callback (Callback): The callback function to execute on the target values.
            item_filter (ItemFilter | None): An optional function to use to filter target
                values before processing them. Defaults to None.
            allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
        """

        callback: Callback[..., Any]
        item_filter: ItemFilter | None = None
        allow_callback_failures: bool = False

    CallbackMapping = dict[type[T] | type[None], list[CallbackDefinition]]

    GetterDefinition = Callable[[T, Any], Any]
    GetterMapping = dict[type[T], GetterDefinition[T]]

    IteratorFactory = Callable[[T], Iterator[Any]]
    IteratorFactoryMapping = dict[type[T], IteratorFactory[T]]

    CallbackMappingInput = dict[
        type[Any] | None,
        Callback[..., Any]
        | CallbackDefinition
        | Collection[
            Callback[..., Any]
            | tuple[Callback[..., Any],]
            | tuple[Callback[..., Any], ItemFilter]
            | tuple[Callback[..., Any], ItemFilter, bool]
            | CallbackDefinition,
        ],
    ]

    class Break(wg_exc.WGUtilitiesError):
        """Escape hatch to allow breaking out of the processing loop from within a callback."""

    def __init__(
        self,
        _cbs: CallbackMappingInput | None = None,
        /,
        *,
        identifier: str = "",
        process_subclasses: bool = True,
        process_type_changes: bool = False,
        process_pydantic_computed_fields: bool = False,
        process_pydantic_extra_fields: bool = False,
        process_pydantic_model_properties: bool = False,
        ignored_loc_lookup_errors: tuple[type[Exception], ...] = (),
    ) -> None:
        """Initialize the JSONProcessor."""
        self.callback_mapping: JSONProcessor.CallbackMapping[Any] = defaultdict(list)
        self.iterator_factory_mapping: JSONProcessor.IteratorFactoryMapping[Any] = {}
        self.getter_mapping: JSONProcessor.GetterMapping[Any] = {}

        self.config = Config(
            process_subclasses=process_subclasses,
            process_type_changes=process_type_changes,
            process_pydantic_computed_fields=process_pydantic_computed_fields,
            process_pydantic_extra_fields=process_pydantic_extra_fields,
            process_pydantic_model_properties=process_pydantic_model_properties,
            ignored_loc_lookup_errors=ignored_loc_lookup_errors,
        )

        self.identifier = identifier or hash(self)

        if _cbs:
            for target_type, cb_val in _cbs.items():
                cb_list: list[JSONProcessor.CallbackDefinition] = []
                if callable(cb_val):
                    # Single callback provided for type
                    cb_list.append(self.cb(cb_val))
                elif isinstance(cb_val, JSONProcessor.CallbackDefinition):
                    # Single CallbackDefinition named tuple provided for type
                    cb_list.append(cb_val)
                elif isinstance(cb_val, Collection):
                    for cb in cb_val:
                        if callable(cb):
                            # Single callbacks provided for type
                            cb_list.append(self.cb(cb))
                        elif isinstance(cb, tuple):
                            # Partial (or full) CallbackDefinition
                            cb_list.append(self.cb(*cb))
                        else:
                            raise InvalidCallbackError(cb, type(cb), callback=cb)
                else:
                    raise InvalidCallbackError(cb_val, type(cb_val), callback=cb_val)

                for cb_def in cb_list:
                    self.register_callback(target_type, cb_def)

        self.processable_types: tuple[type[Any], ...] = (Mapping, Sequence)
        self.unprocessable_types: tuple[type[Any], ...] = (
            (str, bytes) if self.config.process_subclasses else ()
        )

    @staticmethod
    def cb(
        callback: Callback[..., Any],
        item_filter: ItemFilter | None = None,
        allow_callback_failures: bool = False,  # noqa: FBT001,FBT002
        *,
        allow_mutation: bool = True,
    ) -> CallbackDefinition:
        """Create a CallbackDefinition for the given callback.

        Args:
            callback (Callback): The callback function to execute on the target values. Can take None, any, or all
                of the following arguments: `_value_`, `_loc_`, `_obj_type_`, `_depth_`, and any additional keyword
                arguments, which will be passed in from the `JSONProcessor.process` method.
            item_filter (ItemFilter | None): An optional function to use to filter target values before processing
                them. Defaults to None. Function signature is as follows:
                ```python
                def item_filter(item: Any, /, *, loc: K | int | str) -> bool: ...
                ```
            allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
            allow_mutation (bool): Whether the callback is allowed to mutate the input object. Defaults to True.
        """

        if callback.__name__ == "<lambda>":
            callback = JSONProcessor.callback(allow_mutation=allow_mutation)(callback)

        if item_filter and not callable(item_filter):
            raise InvalidItemFilterError(item_filter, callback=callback)

        if not isinstance(allow_callback_failures, bool):
            raise InvalidCallbackError(
                allow_callback_failures,
                type(allow_callback_failures),
                callback=callback,
            )

        return JSONProcessor.CallbackDefinition(
            callback=callback,
            item_filter=item_filter,
            allow_callback_failures=allow_callback_failures,
        )

    def _get_callbacks(
        self,
        typ: type[Any],
    ) -> Generator[CallbackDefinition, None, None]:
        if self.config.process_subclasses:
            for cb_typ in self.callback_mapping:
                if issubclass(typ, cb_typ):
                    yield from self.callback_mapping[cb_typ]
        elif callback_list := self.callback_mapping.get(typ):
            yield from callback_list

    @overload
    def _get_getter_or_iterator_factory(
        self,
        typ: type[T],
        mapping: JSONProcessor.IteratorFactoryMapping[Any],
    ) -> JSONProcessor.IteratorFactory[T] | None: ...

    @overload
    def _get_getter_or_iterator_factory(
        self,
        typ: type[T],
        mapping: JSONProcessor.GetterMapping[Any],
    ) -> JSONProcessor.GetterDefinition[T] | None: ...

    def _get_getter_or_iterator_factory(
        self,
        typ: type[T],
        mapping: JSONProcessor.IteratorFactoryMapping[Any]
        | JSONProcessor.GetterMapping[Any],
    ) -> JSONProcessor.IteratorFactory[T] | JSONProcessor.GetterDefinition[T] | None:
        """Get the getter or iterator for the given type.

        If `config.process_subclasses` is true, the getter or iterator for the first matching
        subclass will be returned. Otherwise, the getter or iterator for the exact type will be
        returned.
        """

        if (exact_match := mapping.get(typ)) or not self.config.process_subclasses:
            return exact_match

        for key, value in mapping.items():
            if issubclass(typ, key):
                return value

        return None

    @overload
    def _get_item(self, obj: BaseModel, loc: str) -> object: ...

    @overload
    def _get_item(self, obj: Mapping[K, object], loc: K) -> object: ...

    @overload
    def _get_item(self, obj: Sequence[object], loc: int) -> object: ...

    @overload
    def _get_item(self, obj: G, loc: object) -> object: ...

    def _get_item(
        self,
        obj: BaseModel | Mapping[K, object] | Sequence[object] | G,
        loc: str | K | int | object,
    ) -> Any | Sentinel:
        with suppress(*self.config.ignored_loc_lookup_errors):
            try:
                if custom_getter := self._get_getter_or_iterator_factory(
                    type(obj), self.getter_mapping
                ):
                    return custom_getter(obj, loc)

                try:
                    return obj[loc]  # type: ignore[index]
                except TypeError as exc:
                    if not isinstance(loc, str) or (
                        not str(exc).endswith("is not subscriptable")
                        and str(exc) != "list indices must be integers or slices, not str"
                    ):
                        raise

                try:
                    return getattr(obj, loc)
                except AttributeError:
                    raise LocNotFoundError(loc, obj) from None

            except LookupError as exc:
                raise LocNotFoundError(loc, obj) from exc

        return JSONProcessor.__SENTINEL

    @staticmethod
    def _set_item(
        obj: Mapping[K, V] | Sequence[object] | BaseModel,
        loc: str | K | int,
        val: V,
    ) -> None:
        with suppress(TypeError):
            obj[loc] = val  # type: ignore[index]
            return

        if not isinstance(loc, str):
            return

        with suppress(AttributeError):
            setattr(obj, loc, val)

        return

    @overload
    def _iterate(self, obj: Mapping[K, Any]) -> Iterator[K]: ...

    @overload
    def _iterate(self, obj: Sequence[Any]) -> Iterator[int]: ...

    @overload
    def _iterate(self, obj: BaseModel) -> Iterator[str]: ...

    def _iterate(
        self,
        obj: Mapping[K, Any] | Sequence[Any] | BaseModel,
    ) -> Iterator[K] | Iterator[int] | Iterator[str] | Sentinel:
        if custom_iter := self._get_getter_or_iterator_factory(
            type(obj), self.iterator_factory_mapping
        ):
            return custom_iter(obj)

        if isinstance(obj, Sequence):
            return iter(range(len(obj)))

        if isinstance(obj, BaseModel):
            iterables: list[Iterable[str]] = [obj.model_fields.keys()]

            if self.config.process_pydantic_model_properties:
                iterables.append(
                    sorted(
                        {
                            name
                            for name, prop in inspect.getmembers(obj.__class__)
                            if isinstance(prop, property)
                            and name not in BASE_MODEL_PROPERTIES
                        },
                    ),
                )

            if self.config.process_pydantic_computed_fields:
                iterables.append(obj.model_computed_fields.keys())

            if self.config.process_pydantic_extra_fields and obj.model_extra:
                iterables.append(obj.model_extra.keys())

            return iter(chain(*iterables))

        try:
            return iter(obj)
        except TypeError as exc:
            if not str(exc).endswith("is not iterable"):
                raise

        return self.__SENTINEL

    @overload
    def _process_item(
        self,
        *,
        obj: BaseModel,
        loc: str,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_item(
        self,
        *,
        obj: Mapping[K, object],
        loc: K,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_item(
        self,
        *,
        obj: Sequence[object],
        loc: int,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None: ...

    def _process_item(
        self,
        *,
        obj: BaseModel | Mapping[K, object] | Sequence[object],
        loc: str | K | int,
        cb: Callback[..., Any],
        depth: int,
        orig_item_type: type[Any],
        kwargs: dict[str, Any],
    ) -> None:
        try:
            out = cb(
                _value_=self._get_item(obj, loc),
                _loc_=loc,
                _obj_type_=type(obj),
                _depth_=depth,
                **kwargs,
            )
        except TypeError as exc:
            arg_error: type[InvalidCallbackArgumentsError]
            for arg_error in InvalidCallbackArgumentsError.subclasses():  # type: ignore[assignment]
                if match := arg_error.PATTERN.search(str(exc)):
                    raise arg_error(*match.groups(), callback=cb) from None

            raise

        if out is not self.__SENTINEL:
            self._set_item(obj, loc, out)

            if self.config.process_type_changes and type(out) != orig_item_type:
                self._process_loc(
                    obj=obj,  # type: ignore[arg-type]
                    loc=loc,  # type: ignore[arg-type]
                    depth=depth,
                    kwargs=kwargs,
                )

    @overload
    def _process_loc(
        self,
        *,
        obj: BaseModel,
        loc: str,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_loc(
        self,
        *,
        obj: Sequence[object],
        loc: int,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None: ...

    @overload
    def _process_loc(
        self,
        *,
        obj: Mapping[K, object],
        loc: K,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None: ...

    def _process_loc(
        self,
        *,
        obj: BaseModel | Mapping[K, object] | Sequence[object],
        loc: str | K | int,
        depth: int,
        kwargs: dict[str, Any],
    ) -> None:
        try:
            item = self._get_item(obj, loc)
        except LocNotFoundError:
            return

        for cb, item_filter, allow_failures in self._get_callbacks(
            orig_item_type := type(item),
        ):
            if not item_filter or bool(item_filter(item, loc=loc)):
                try:
                    self._process_item(
                        obj=obj,  # type: ignore[arg-type]
                        loc=loc,  # type: ignore[arg-type]
                        cb=cb,
                        depth=depth,
                        orig_item_type=orig_item_type,
                        kwargs=kwargs,
                    )
                except (self.Break, InvalidCallbackArgumentsError):
                    raise
                except Exception:
                    if not allow_failures:
                        raise

    def process(
        self,
        obj: Mapping[K, object] | Sequence[object],
        /,
        __depth: int = 0,
        __processed_models: set[BaseModel] | None = None,
        **kwargs: Any,
    ) -> None:
        """Recursively process a JSON object with the registered callbacks.

        Args:
            obj: The JSON object to process.
            kwargs: Any additional keyword arguments to pass to the callback(s).
        """

        for loc in self._iterate(obj):
            try:
                self._process_loc(
                    obj=obj,
                    loc=loc,
                    depth=__depth,
                    kwargs=kwargs,
                )
            except self.Break:
                break

            item = self._get_item(obj, loc)

            if isinstance(item, self.processable_types) and not isinstance(
                item,
                self.unprocessable_types,
            ):
                self.process(
                    item,
                    _JSONProcessor__depth=__depth + 1,
                    _JSONProcessor__processed_models=__processed_models,
                    **kwargs,
                )
            elif (
                PYDANTIC_INSTALLED
                and isinstance(item, BaseModel)
                and item not in (__processed_models or set())
            ):
                self.process_model(
                    item,
                    _JSONProcessor__depth=__depth + 1,
                    _JSONProcessor__processed_models=__processed_models,
                    **kwargs,
                )

    def process_anything(self, obj: Any, **kwargs: Any) -> None:  # pragma: no cover
        """Process anything that can be processed."""
        self.process(obj, **kwargs)

    if PYDANTIC_INSTALLED:

        def process_model(
            self,
            model: BaseModel,
            /,
            __depth: int = 0,
            __processed_models: set[BaseModel] | None = None,
            **kwargs: Any,
        ) -> None:
            """Recursively process a Pydantic model with the registered callbacks.

            Args:
                model: The Pydantic model to process.
                kwargs: Any additional keyword arguments to pass to the callback(s).
            """
            if __processed_models is None:
                __processed_models = {model}
            else:
                __processed_models.add(model)

            for loc in self._iterate(model):
                try:
                    self._process_loc(obj=model, loc=loc, depth=__depth, kwargs=kwargs)
                except self.Break:
                    break

                try:
                    item = getattr(model, loc)
                except self.config.ignored_loc_lookup_errors:
                    continue

                if isinstance(item, BaseModel) and item not in __processed_models:
                    self.process_model(
                        item,
                        _JSONProcessor__depth=__depth + 1,
                        _JSONProcessor__processed_models=__processed_models,
                        **kwargs,
                    )
                elif isinstance(item, Mapping | Sequence) and not isinstance(
                    item,
                    (str, bytes),
                ):
                    self.process(
                        item,
                        _JSONProcessor__depth=__depth + 1,
                        _JSONProcessor__processed_models=__processed_models,
                        **kwargs,
                    )

    def register_callback(
        self,
        target_type: type[C] | None,
        callback_def: CallbackDefinition,
    ) -> None:
        """Register a new callback for use when processing any JSON objects.

        Args:
            target_type (type): The type of the values to be processed.
            callback_def (CallbackDefinition): The callback definition to register.
        """
        decorated = (
            callback_def.callback.__func__
            if inspect.ismethod(callback_def.callback)
            else callback_def.callback
        )

        if decorated not in JSONProcessor._DECORATED_CALLBACKS:
            raise CallbackNotDecoratedError(decorated)

        self.callback_mapping[target_type or type(None)].append(callback_def)

    def register_custom_getter(
        self,
        target_type: type[G],
        getter: GetterDefinition[G],
        *,
        add_to_processable_type_list: bool = True,
    ) -> None:
        """Register a custom getter for use when processing any JSON objects.

        Args:
            target_type (type): The type of the values to be processed.
            getter (Callable): The custom getter to register.
            add_to_processable_type_list (bool): Whether to add the target type to the list of
                processable types. Defaults to True.
        """
        self.getter_mapping[target_type] = getter

        if add_to_processable_type_list and target_type not in self.processable_types:
            self.processable_types = (*self.processable_types, target_type)

    def register_custom_iterator(
        self,
        target_type: type[I],
        iterator: IteratorFactory[I],
        *,
        add_to_processable_type_list: bool = True,
    ) -> None:
        """Register a custom iterator for use when processing any JSON objects.

        Args:
            target_type (type): The type of the values to be processed.
            iterator (Callable): The custom iterator to register.
            add_to_processable_type_list (bool): Whether to add the target type to the list of
                processable types. Defaults to True.
        """
        self.iterator_factory_mapping[target_type] = iterator

        if add_to_processable_type_list and target_type not in self.processable_types:
            self.processable_types = (*self.processable_types, target_type)

    @overload
    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: Literal[True] = True,
    ) -> Callable[[Callable[P, R]], Callback[P, R]]: ...

    @overload
    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: Literal[False],
    ) -> Callable[[Callable[P, R]], Callback[P, Sentinel]]: ...

    @overload
    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: bool = ...,
    ) -> Callable[[Callable[P, R]], Callback[P, R | Sentinel]]: ...

    @classmethod
    def callback(
        cls,
        *,
        allow_mutation: bool = True,
    ) -> Callable[[Callable[P, R]], Callback[P, R | Sentinel]]:
        """Decorator to mark a function as a callback for use with the JSONProcessor.

        Warning:
            `allow_mutation` only blocks the return value from being used to update the input
            object. It does not prevent the callback from mutating the input object (or other
            objects passed in as arguments) in place.

        Args:
            allow_mutation (bool): Whether the callback is allowed to mutate the input object.
                Defaults to True.
        """

        def _decorator(func: Callable[P, R]) -> Callback[P, R | Sentinel]:
            """Decorator to mark a function as a callback for use with the JSONProcessor."""

            if isinstance(func, classmethod):
                raise InvalidCallbackError(
                    "@JSONProcessor.callback must be used _after_ @classmethod",
                    callback=func,
                )

            arg_names, kwarg_names = [], []

            for name, param in inspect.signature(func).parameters.items():
                if param.kind == param.POSITIONAL_ONLY:
                    arg_names.append(name)
                else:
                    kwarg_names.append(name)

            def filter_kwargs(
                kwargs: dict[str, Any],
                /,
            ) -> tuple[list[Any], dict[str, Any]]:
                a = []
                for an in arg_names:
                    with suppress(KeyError):
                        a.append(kwargs[an])

                kw = {}
                for kn in kwarg_names:
                    with suppress(KeyError):
                        kw[kn] = kwargs[kn]

                return a, kw

            # This is the same `cb` called in the `JSONProcessor._process_loc` method - no positional
            # arguments are explicitly passed in (and would be rejected by `JSONProcessor.process` anyway),
            # unless the callback is a bound method. In this case, the `cls`/`self` argument is passed in
            # as the first positional argument, hence the need for `*bound_args` below

            if allow_mutation:

                @wraps(func)
                def cb(*bound_args: P.args, **process_kwargs: P.kwargs) -> R:
                    args, kwargs = filter_kwargs(process_kwargs)
                    return func(*bound_args, *args, **kwargs)

            else:

                @wraps(func)
                def cb(
                    *bound_args: P.args,
                    **process_kwargs: P.kwargs,
                ) -> Sentinel:
                    args, kwargs = filter_kwargs(process_kwargs)
                    func(*bound_args, *args, **kwargs)
                    return JSONProcessor.__SENTINEL

            cls._DECORATED_CALLBACKS.add(cb)

            return cb

        return _decorator
Break

Bases: WGUtilitiesError

Escape hatch to allow breaking out of the processing loop from within a callback.

Source code in wg_utilities/helpers/processor/json.py 
237
238
class Break(wg_exc.WGUtilitiesError):
    """Escape hatch to allow breaking out of the processing loop from within a callback."""
CallbackDefinition

Bases: NamedTuple

A named tuple to hold the callback function and its associated data.

Attributes:

Name Type Description
callback Callback

The callback function to execute on the target values.
item_filter ItemFilter | None

An optional function to use to filter target values before processing them. Defaults to None.
allow_callback_failures bool

Whether to allow callback failures. Defaults to False.
Source code in wg_utilities/helpers/processor/json.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
class CallbackDefinition(NamedTuple):
    """A named tuple to hold the callback function and its associated data.

    Attributes:
        callback (Callback): The callback function to execute on the target values.
        item_filter (ItemFilter | None): An optional function to use to filter target
            values before processing them. Defaults to None.
        allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
    """

    callback: Callback[..., Any]
    item_filter: ItemFilter | None = None
    allow_callback_failures: bool = False
__init__(_cbs=None, /, *, identifier='', process_subclasses=True, process_type_changes=False, process_pydantic_computed_fields=False, process_pydantic_extra_fields=False, process_pydantic_model_properties=False, ignored_loc_lookup_errors=())

Initialize the JSONProcessor.

Source code in wg_utilities/helpers/processor/json.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def __init__(
    self,
    _cbs: CallbackMappingInput | None = None,
    /,
    *,
    identifier: str = "",
    process_subclasses: bool = True,
    process_type_changes: bool = False,
    process_pydantic_computed_fields: bool = False,
    process_pydantic_extra_fields: bool = False,
    process_pydantic_model_properties: bool = False,
    ignored_loc_lookup_errors: tuple[type[Exception], ...] = (),
) -> None:
    """Initialize the JSONProcessor."""
    self.callback_mapping: JSONProcessor.CallbackMapping[Any] = defaultdict(list)
    self.iterator_factory_mapping: JSONProcessor.IteratorFactoryMapping[Any] = {}
    self.getter_mapping: JSONProcessor.GetterMapping[Any] = {}

    self.config = Config(
        process_subclasses=process_subclasses,
        process_type_changes=process_type_changes,
        process_pydantic_computed_fields=process_pydantic_computed_fields,
        process_pydantic_extra_fields=process_pydantic_extra_fields,
        process_pydantic_model_properties=process_pydantic_model_properties,
        ignored_loc_lookup_errors=ignored_loc_lookup_errors,
    )

    self.identifier = identifier or hash(self)

    if _cbs:
        for target_type, cb_val in _cbs.items():
            cb_list: list[JSONProcessor.CallbackDefinition] = []
            if callable(cb_val):
                # Single callback provided for type
                cb_list.append(self.cb(cb_val))
            elif isinstance(cb_val, JSONProcessor.CallbackDefinition):
                # Single CallbackDefinition named tuple provided for type
                cb_list.append(cb_val)
            elif isinstance(cb_val, Collection):
                for cb in cb_val:
                    if callable(cb):
                        # Single callbacks provided for type
                        cb_list.append(self.cb(cb))
                    elif isinstance(cb, tuple):
                        # Partial (or full) CallbackDefinition
                        cb_list.append(self.cb(*cb))
                    else:
                        raise InvalidCallbackError(cb, type(cb), callback=cb)
            else:
                raise InvalidCallbackError(cb_val, type(cb_val), callback=cb_val)

            for cb_def in cb_list:
                self.register_callback(target_type, cb_def)

    self.processable_types: tuple[type[Any], ...] = (Mapping, Sequence)
    self.unprocessable_types: tuple[type[Any], ...] = (
        (str, bytes) if self.config.process_subclasses else ()
    )
callback(*, allow_mutation=True) classmethod

Decorator to mark a function as a callback for use with the JSONProcessor.

Warning

allow_mutation only blocks the return value from being used to update the input object. It does not prevent the callback from mutating the input object (or other objects passed in as arguments) in place.

Parameters:

Name Type Description Default
allow_mutation bool

Whether the callback is allowed to mutate the input object. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
@classmethod
def callback(
    cls,
    *,
    allow_mutation: bool = True,
) -> Callable[[Callable[P, R]], Callback[P, R | Sentinel]]:
    """Decorator to mark a function as a callback for use with the JSONProcessor.

    Warning:
        `allow_mutation` only blocks the return value from being used to update the input
        object. It does not prevent the callback from mutating the input object (or other
        objects passed in as arguments) in place.

    Args:
        allow_mutation (bool): Whether the callback is allowed to mutate the input object.
            Defaults to True.
    """

    def _decorator(func: Callable[P, R]) -> Callback[P, R | Sentinel]:
        """Decorator to mark a function as a callback for use with the JSONProcessor."""

        if isinstance(func, classmethod):
            raise InvalidCallbackError(
                "@JSONProcessor.callback must be used _after_ @classmethod",
                callback=func,
            )

        arg_names, kwarg_names = [], []

        for name, param in inspect.signature(func).parameters.items():
            if param.kind == param.POSITIONAL_ONLY:
                arg_names.append(name)
            else:
                kwarg_names.append(name)

        def filter_kwargs(
            kwargs: dict[str, Any],
            /,
        ) -> tuple[list[Any], dict[str, Any]]:
            a = []
            for an in arg_names:
                with suppress(KeyError):
                    a.append(kwargs[an])

            kw = {}
            for kn in kwarg_names:
                with suppress(KeyError):
                    kw[kn] = kwargs[kn]

            return a, kw

        # This is the same `cb` called in the `JSONProcessor._process_loc` method - no positional
        # arguments are explicitly passed in (and would be rejected by `JSONProcessor.process` anyway),
        # unless the callback is a bound method. In this case, the `cls`/`self` argument is passed in
        # as the first positional argument, hence the need for `*bound_args` below

        if allow_mutation:

            @wraps(func)
            def cb(*bound_args: P.args, **process_kwargs: P.kwargs) -> R:
                args, kwargs = filter_kwargs(process_kwargs)
                return func(*bound_args, *args, **kwargs)

        else:

            @wraps(func)
            def cb(
                *bound_args: P.args,
                **process_kwargs: P.kwargs,
            ) -> Sentinel:
                args, kwargs = filter_kwargs(process_kwargs)
                func(*bound_args, *args, **kwargs)
                return JSONProcessor.__SENTINEL

        cls._DECORATED_CALLBACKS.add(cb)

        return cb

    return _decorator
cb(callback, item_filter=None, allow_callback_failures=False, *, allow_mutation=True) staticmethod

Create a CallbackDefinition for the given callback.

Parameters:

Name Type Description Default
callback Callback

The callback function to execute on the target values. Can take None, any, or all of the following arguments: _value_, _loc_, _obj_type_, _depth_, and any additional keyword arguments, which will be passed in from the JSONProcessor.process method.

 required
item_filter ItemFilter | None

An optional function to use to filter target values before processing them. Defaults to None. Function signature is as follows:

def item_filter(item: Any, /, *, loc: K | int | str) -> bool: ...
None
allow_callback_failures bool

Whether to allow callback failures. Defaults to False.

 False
allow_mutation bool

Whether the callback is allowed to mutate the input object. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
@staticmethod
def cb(
    callback: Callback[..., Any],
    item_filter: ItemFilter | None = None,
    allow_callback_failures: bool = False,  # noqa: FBT001,FBT002
    *,
    allow_mutation: bool = True,
) -> CallbackDefinition:
    """Create a CallbackDefinition for the given callback.

    Args:
        callback (Callback): The callback function to execute on the target values. Can take None, any, or all
            of the following arguments: `_value_`, `_loc_`, `_obj_type_`, `_depth_`, and any additional keyword
            arguments, which will be passed in from the `JSONProcessor.process` method.
        item_filter (ItemFilter | None): An optional function to use to filter target values before processing
            them. Defaults to None. Function signature is as follows:
            ```python
            def item_filter(item: Any, /, *, loc: K | int | str) -> bool: ...
            ```
        allow_callback_failures (bool): Whether to allow callback failures. Defaults to False.
        allow_mutation (bool): Whether the callback is allowed to mutate the input object. Defaults to True.
    """

    if callback.__name__ == "<lambda>":
        callback = JSONProcessor.callback(allow_mutation=allow_mutation)(callback)

    if item_filter and not callable(item_filter):
        raise InvalidItemFilterError(item_filter, callback=callback)

    if not isinstance(allow_callback_failures, bool):
        raise InvalidCallbackError(
            allow_callback_failures,
            type(allow_callback_failures),
            callback=callback,
        )

    return JSONProcessor.CallbackDefinition(
        callback=callback,
        item_filter=item_filter,
        allow_callback_failures=allow_callback_failures,
    )
process(obj, /, __depth=0, __processed_models=None, **kwargs)

Recursively process a JSON object with the registered callbacks.

Parameters:

Name Type Description Default
obj Mapping[K, object] | Sequence[object]

The JSON object to process.

 required
kwargs Any

Any additional keyword arguments to pass to the callback(s).

 {}
Source code in wg_utilities/helpers/processor/json.py 
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
def process(
    self,
    obj: Mapping[K, object] | Sequence[object],
    /,
    __depth: int = 0,
    __processed_models: set[BaseModel] | None = None,
    **kwargs: Any,
) -> None:
    """Recursively process a JSON object with the registered callbacks.

    Args:
        obj: The JSON object to process.
        kwargs: Any additional keyword arguments to pass to the callback(s).
    """

    for loc in self._iterate(obj):
        try:
            self._process_loc(
                obj=obj,
                loc=loc,
                depth=__depth,
                kwargs=kwargs,
            )
        except self.Break:
            break

        item = self._get_item(obj, loc)

        if isinstance(item, self.processable_types) and not isinstance(
            item,
            self.unprocessable_types,
        ):
            self.process(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )
        elif (
            PYDANTIC_INSTALLED
            and isinstance(item, BaseModel)
            and item not in (__processed_models or set())
        ):
            self.process_model(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )
process_anything(obj, **kwargs)

Process anything that can be processed.

Source code in wg_utilities/helpers/processor/json.py 
686
687
688
def process_anything(self, obj: Any, **kwargs: Any) -> None:  # pragma: no cover
    """Process anything that can be processed."""
    self.process(obj, **kwargs)
process_model(model, /, __depth=0, __processed_models=None, **kwargs)

Recursively process a Pydantic model with the registered callbacks.

Parameters:

Name Type Description Default
model BaseModel

The Pydantic model to process.

 required
kwargs Any

Any additional keyword arguments to pass to the callback(s).

 {}
Source code in wg_utilities/helpers/processor/json.py 
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
def process_model(
    self,
    model: BaseModel,
    /,
    __depth: int = 0,
    __processed_models: set[BaseModel] | None = None,
    **kwargs: Any,
) -> None:
    """Recursively process a Pydantic model with the registered callbacks.

    Args:
        model: The Pydantic model to process.
        kwargs: Any additional keyword arguments to pass to the callback(s).
    """
    if __processed_models is None:
        __processed_models = {model}
    else:
        __processed_models.add(model)

    for loc in self._iterate(model):
        try:
            self._process_loc(obj=model, loc=loc, depth=__depth, kwargs=kwargs)
        except self.Break:
            break

        try:
            item = getattr(model, loc)
        except self.config.ignored_loc_lookup_errors:
            continue

        if isinstance(item, BaseModel) and item not in __processed_models:
            self.process_model(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )
        elif isinstance(item, Mapping | Sequence) and not isinstance(
            item,
            (str, bytes),
        ):
            self.process(
                item,
                _JSONProcessor__depth=__depth + 1,
                _JSONProcessor__processed_models=__processed_models,
                **kwargs,
            )
register_callback(target_type, callback_def)

Register a new callback for use when processing any JSON objects.

Parameters:

Name Type Description Default
target_type type

The type of the values to be processed.

 required
callback_def CallbackDefinition

The callback definition to register.

 required
Source code in wg_utilities/helpers/processor/json.py 
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
def register_callback(
    self,
    target_type: type[C] | None,
    callback_def: CallbackDefinition,
) -> None:
    """Register a new callback for use when processing any JSON objects.

    Args:
        target_type (type): The type of the values to be processed.
        callback_def (CallbackDefinition): The callback definition to register.
    """
    decorated = (
        callback_def.callback.__func__
        if inspect.ismethod(callback_def.callback)
        else callback_def.callback
    )

    if decorated not in JSONProcessor._DECORATED_CALLBACKS:
        raise CallbackNotDecoratedError(decorated)

    self.callback_mapping[target_type or type(None)].append(callback_def)
register_custom_getter(target_type, getter, *, add_to_processable_type_list=True)

Register a custom getter for use when processing any JSON objects.

Parameters:

Name Type Description Default
target_type type

The type of the values to be processed.

 required
getter Callable

The custom getter to register.

 required
add_to_processable_type_list bool

Whether to add the target type to the list of processable types. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
def register_custom_getter(
    self,
    target_type: type[G],
    getter: GetterDefinition[G],
    *,
    add_to_processable_type_list: bool = True,
) -> None:
    """Register a custom getter for use when processing any JSON objects.

    Args:
        target_type (type): The type of the values to be processed.
        getter (Callable): The custom getter to register.
        add_to_processable_type_list (bool): Whether to add the target type to the list of
            processable types. Defaults to True.
    """
    self.getter_mapping[target_type] = getter

    if add_to_processable_type_list and target_type not in self.processable_types:
        self.processable_types = (*self.processable_types, target_type)
register_custom_iterator(target_type, iterator, *, add_to_processable_type_list=True)

Register a custom iterator for use when processing any JSON objects.

Parameters:

Name Type Description Default
target_type type

The type of the values to be processed.

 required
iterator Callable

The custom iterator to register.

 required
add_to_processable_type_list bool

Whether to add the target type to the list of processable types. Defaults to True.

 True
Source code in wg_utilities/helpers/processor/json.py 
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
def register_custom_iterator(
    self,
    target_type: type[I],
    iterator: IteratorFactory[I],
    *,
    add_to_processable_type_list: bool = True,
) -> None:
    """Register a custom iterator for use when processing any JSON objects.

    Args:
        target_type (type): The type of the values to be processed.
        iterator (Callable): The custom iterator to register.
        add_to_processable_type_list (bool): Whether to add the target type to the list of
            processable types. Defaults to True.
    """
    self.iterator_factory_mapping[target_type] = iterator

    if add_to_processable_type_list and target_type not in self.processable_types:
        self.processable_types = (*self.processable_types, target_type)

LocNotFoundError

Bases: BadUsageError, LookupError

Raised when a location is not found in the object.

Source code in wg_utilities/helpers/processor/json.py 
66
67
68
69
70
71
72
73
74
75
class LocNotFoundError(wg_exc.BadUsageError, LookupError):
    """Raised when a location is not found in the object."""

    def __init__(
        self,
        loc: Any,
        /,
        obj: Any,
    ) -> None:
        super().__init__(f"Location {loc!r} not found in object {obj!r}.")

MissingArgError

Bases: InvalidCallbackArgumentsError

Raised when a required argument is missing.

Source code in wg_utilities/helpers/processor/json.py 
100
101
102
103
104
class MissingArgError(InvalidCallbackArgumentsError):
    """Raised when a required argument is missing."""

    ARG_TYPE = "positional"
    PATTERN = re.compile(rf"missing \d+ required {ARG_TYPE} argument: '(.+)'")

MissingKwargError

Bases: InvalidCallbackArgumentsError

Raised when a required keyword-only argument is missing.

Source code in wg_utilities/helpers/processor/json.py 
107
108
109
110
111
class MissingKwargError(InvalidCallbackArgumentsError):
    """Raised when a required keyword-only argument is missing."""

    ARG_TYPE = "keyword-only"
    PATTERN = re.compile(rf"missing \d+ required {ARG_TYPE} argument: '(.+)'")