pedantic

  1from pedantic.decorators import (
  2    calculate_in_subprocess,
  3    decorate_class,
  4    deprecated,
  5    frozen_dataclass,
  6    frozen_type_safe_dataclass,
  7    in_subprocess,
  8    overrides,
  9    pedantic,
 10    retry,
 11    safe_async_contextmanager,
 12    safe_contextmanager,
 13    trace,
 14)
 15from pedantic.decorators.validate.exceptions import (
 16    ConversionError,
 17    ExceptionDictKey,
 18    InvalidHeader,
 19    ParameterException,
 20    TooManyArguments,
 21    ValidateException,
 22    ValidatorException,
 23)
 24from pedantic.decorators.validate.parameters import (
 25    Deserializable,
 26    EnvironmentVariableParameter,
 27    ExternalParameter,
 28    Parameter,
 29)
 30from pedantic.decorators.validate.validate import (
 31    ReturnAs,
 32    validate,
 33)
 34from pedantic.decorators.validate.validators import (
 35    DatetimeIsoFormat,
 36    DateTimeUnixTimestamp,
 37    Email,
 38    ForEach,
 39    IsEnum,
 40    IsUuid,
 41    MatchPattern,
 42    Max,
 43    MaxLength,
 44    Min,
 45    MinLength,
 46    NotEmpty,
 47    Validator,
 48)
 49from pedantic.helper_fn import run_doctest_of_single_function
 50from pedantic.mixins import (
 51    DecoratorType,
 52    GenericMixin,
 53    WithDecoratedMethods,
 54    create_decorator,
 55)
 56from pedantic.type_checking_logic import (
 57    assert_value_matches_type,
 58    resolve_forward_ref,
 59)
 60
 61__all__ = [
 62    'ConversionError',
 63    'DateTimeUnixTimestamp',
 64    'DatetimeIsoFormat',
 65    'DecoratorType',
 66    'Deserializable',
 67    'Email',
 68    'EnvironmentVariableParameter',
 69    'ExceptionDictKey',
 70    'ExternalParameter',
 71    'ExternalParameter',
 72    'ForEach',
 73    'GenericMixin',
 74    'InvalidHeader',
 75    'IsEnum',
 76    'IsUuid',
 77    'MatchPattern',
 78    'Max',
 79    'MaxLength',
 80    'Min',
 81    'MinLength',
 82    'NotEmpty',
 83    'Parameter',
 84    'ParameterException',
 85    'ReturnAs',
 86    'TooManyArguments',
 87    'ValidateException',
 88    'Validator',
 89    'ValidatorException',
 90    'WithDecoratedMethods',
 91    'assert_value_matches_type',
 92    'calculate_in_subprocess',
 93    'create_decorator',
 94    'decorate_class',
 95    'deprecated',
 96    'frozen_dataclass',
 97    'frozen_type_safe_dataclass',
 98    'in_subprocess',
 99    'overrides',
100    'pedantic',
101    'resolve_forward_ref',
102    'retry',
103    'run_doctest_of_single_function',
104    'safe_async_contextmanager',
105    'safe_contextmanager',
106    'trace',
107    'validate',
108]
109
110try:
111    from pedantic.decorators.validate.parameters import (
112        FlaskFormParameter,
113        FlaskGetParameter,
114        FlaskHeaderParameter,
115        FlaskJsonParameter,
116        FlaskParameter,
117        FlaskPathParameter,
118        GenericFlaskDeserializer,
119    )
120
121    __all__ +=[
122        'FlaskFormParameter',
123        'FlaskGetParameter',
124        'FlaskHeaderParameter',
125        'FlaskJsonParameter',
126        'FlaskParameter',
127        'FlaskPathParameter',
128        'GenericFlaskDeserializer',
129    ]
130except ImportError:
131    pass  # no Flask installed
class ConversionError(pedantic.ValidateException):
78class ConversionError(ValidateException):
79    """Is raised if a type cast failed."""

Is raised if a type cast failed.

class DateTimeUnixTimestamp(pedantic.Validator):
 8class DateTimeUnixTimestamp(Validator): # noqa: D101
 9    @overrides(Validator)
10    def validate(self, value: float | str) -> datetime: # noqa: D102
11        if not isinstance(value, (int, float, str)):
12            self.raise_exception(msg=f'Invalid seconds since 1970: {value}', value=value)
13
14        try:
15            seconds = float(value)
16        except ValueError:
17            return self.raise_exception(msg=f'Could parse {value} to float.', value=value)
18
19        try:
20            return datetime(year=1970, month=1, day=1, tzinfo=UTC) + timedelta(seconds=seconds)
21        except OverflowError:
22            return self.raise_exception(
23                msg=f'Date value out of range. Make sure you send SECONDS since 1970. Got: {value}', value=value)

Base class for validator classes.

@overrides(Validator)
def validate(self, value: float | str) -> datetime.datetime:
 9    @overrides(Validator)
10    def validate(self, value: float | str) -> datetime: # noqa: D102
11        if not isinstance(value, (int, float, str)):
12            self.raise_exception(msg=f'Invalid seconds since 1970: {value}', value=value)
13
14        try:
15            seconds = float(value)
16        except ValueError:
17            return self.raise_exception(msg=f'Could parse {value} to float.', value=value)
18
19        try:
20            return datetime(year=1970, month=1, day=1, tzinfo=UTC) + timedelta(seconds=seconds)
21        except OverflowError:
22            return self.raise_exception(
23                msg=f'Date value out of range. Make sure you send SECONDS since 1970. Got: {value}', value=value)

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class DatetimeIsoFormat(pedantic.Validator):
 8class DatetimeIsoFormat(Validator): # noqa: D101
 9    @overrides(Validator)
10    def validate(self, value: str) -> datetime: # noqa: D102
11        try:
12            value = datetime.fromisoformat(value)
13        except (TypeError, ValueError, AttributeError):
14            self.raise_exception(msg=f'invalid value: {value} is not a datetime in ISO format', value=value)
15
16        return value

Base class for validator classes.

@overrides(Validator)
def validate(self, value: str) -> datetime.datetime:
 9    @overrides(Validator)
10    def validate(self, value: str) -> datetime: # noqa: D102
11        try:
12            value = datetime.fromisoformat(value)
13        except (TypeError, ValueError, AttributeError):
14            self.raise_exception(msg=f'invalid value: {value} is not a datetime in ISO format', value=value)
15
16        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class DecoratorType(enum.StrEnum):
10class DecoratorType(StrEnum):
11    """
12    The interface that defines all possible decorators types.
13
14    The values of this enum are used as property names and the properties are added to the decorated functions.
15    So I would recommend naming them with a leading underscore to keep them private and also write it lowercase.
16
17    Example:
18        >>> class Decorators(DecoratorType):
19        ...     FOO = '_foo'
20    """

The interface that defines all possible decorators types.

The values of this enum are used as property names and the properties are added to the decorated functions. So I would recommend naming them with a leading underscore to keep them private and also write it lowercase.

Example:

class Decorators(DecoratorType): ... FOO = '_foo'

class Deserializable(abc.ABC):
 6class Deserializable(ABC):
 7    """A tiny interface which has a static from_json() method which acts like a named constructor."""
 8
 9    @staticmethod
10    @abstractmethod
11    def from_json(data: dict[str, Any]) -> 'Deserializable':
12        """A named constructor which creates an object from JSON."""

A tiny interface which has a static from_json() method which acts like a named constructor.

@staticmethod
@abstractmethod
def from_json( data: dict[str, typing.Any]) -> Deserializable:
 9    @staticmethod
10    @abstractmethod
11    def from_json(data: dict[str, Any]) -> 'Deserializable':
12        """A named constructor which creates an object from JSON."""

A named constructor which creates an object from JSON.

class Email(pedantic.Validator):
11class Email(Validator): # noqa: D101
12    def __init__(self, email_pattern: str = REGEX_EMAIL, post_processor: Callable[[str], str] = lambda x: x) -> None: # noqa: D107
13        self._pattern = email_pattern
14        self._post_processor = post_processor
15
16    @overrides(Validator)
17    def validate(self, value: str) -> str: # noqa: D102
18        if not re.fullmatch(pattern=self._pattern, string=value):
19            self.raise_exception(msg=f'invalid email address: {value}', value=value)
20
21        return self._post_processor(value)

Base class for validator classes.

Email( email_pattern: str = '[^@\\s]+@[^@\\s]+\\.[a-zA-Z0-9]+$', post_processor: Callable[[str], str] = <function Email.<lambda>>)
12    def __init__(self, email_pattern: str = REGEX_EMAIL, post_processor: Callable[[str], str] = lambda x: x) -> None: # noqa: D107
13        self._pattern = email_pattern
14        self._post_processor = post_processor
@overrides(Validator)
def validate(self, value: str) -> str:
16    @overrides(Validator)
17    def validate(self, value: str) -> str: # noqa: D102
18        if not re.fullmatch(pattern=self._pattern, string=value):
19            self.raise_exception(msg=f'invalid email address: {value}', value=value)
20
21        return self._post_processor(value)

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class EnvironmentVariableParameter(pedantic.ExternalParameter):
12class EnvironmentVariableParameter(ExternalParameter):   # noqa: D101
13    def __init__(  # noqa: D107, PLR0913
14        self,
15        name: str,
16        env_var_name: str | None = None,
17        value_type: type[str | bool | int | float] = str,
18        validators: Iterable[Validator] | None = None,
19        required: bool = True,
20default: Any = NoValue,
21     ) -> None:
22        super().__init__(name=name, validators=validators, default=default, value_type=value_type, required=required)
23
24        if value_type not in [str, bool, int, float]:
25            raise AssertionError('value_type needs to be one of these: str, bool, int & float')
26
27        if env_var_name is None:
28            self._env_var_name = name
29        else:
30            self._env_var_name = env_var_name
31
32    @overrides(ExternalParameter)
33    def has_value(self) -> bool:   # noqa: D102
34        return self._env_var_name in os.environ
35
36    @overrides(ExternalParameter)
37    def load_value(self) -> Any:   # noqa: D102
38        return os.environ[self._env_var_name].strip()

The interface for all external parameters.

EnvironmentVariableParameter( name: str, env_var_name: str | None = None, value_type: type[str | bool | int | float] = <class 'str'>, validators: Iterable[Validator] | None = None, required: bool = True, default: Any = <class 'pedantic.decorators.validate.parameters.abstract_parameter.NoValue'>)
13    def __init__(  # noqa: D107, PLR0913
14        self,
15        name: str,
16        env_var_name: str | None = None,
17        value_type: type[str | bool | int | float] = str,
18        validators: Iterable[Validator] | None = None,
19        required: bool = True,
20default: Any = NoValue,
21     ) -> None:
22        super().__init__(name=name, validators=validators, default=default, value_type=value_type, required=required)
23
24        if value_type not in [str, bool, int, float]:
25            raise AssertionError('value_type needs to be one of these: str, bool, int & float')
26
27        if env_var_name is None:
28            self._env_var_name = name
29        else:
30            self._env_var_name = env_var_name
@overrides(ExternalParameter)
def has_value(self) -> bool:
32    @overrides(ExternalParameter)
33    def has_value(self) -> bool:   # noqa: D102
34        return self._env_var_name in os.environ

Returns True if the value can be fetched.

@overrides(ExternalParameter)
def load_value(self) -> Any:
36    @overrides(ExternalParameter)
37    def load_value(self) -> Any:   # noqa: D102
38        return os.environ[self._env_var_name].strip()

Loads a value and returns it.

class ExceptionDictKey:
5class ExceptionDictKey: # noqa: D101
6    VALUE = 'VALUE'
7    MESSAGE = 'MESSAGE'
8    PARAMETER = 'PARAMETER'
9    VALIDATOR = 'VALIDATOR'
VALUE = 'VALUE'
MESSAGE = 'MESSAGE'
PARAMETER = 'PARAMETER'
VALIDATOR = 'VALIDATOR'
class ExternalParameter(pedantic.Parameter, abc.ABC):
 8class ExternalParameter(Parameter, ABC):
 9    """The interface for all external parameters."""
10
11    @abstractmethod
12    def has_value(self) -> bool:
13        """Returns True if the value can be fetched."""
14
15    @abstractmethod
16    def load_value(self) -> Any:
17        """Loads a value and returns it."""

The interface for all external parameters.

@abstractmethod
def has_value(self) -> bool:
11    @abstractmethod
12    def has_value(self) -> bool:
13        """Returns True if the value can be fetched."""

Returns True if the value can be fetched.

@abstractmethod
def load_value(self) -> Any:
15    @abstractmethod
16    def load_value(self) -> Any:
17        """Loads a value and returns it."""

Loads a value and returns it.

class ForEach(pedantic.Validator):
10class ForEach(Validator):  # noqa: D101
11    def __init__(self, validators: Validator | Iterable[Validator]) -> None:  # noqa: D107
12        if isinstance(validators, Validator):
13            self._validators = [validators]
14        else:
15            self._validators = validators
16
17    @overrides(Validator)
18    def validate(self, value: Iterable[Any]) -> list[Any]:  # noqa: D102
19        if not isinstance(value, collections.abc.Iterable):
20            self.raise_exception(msg=f'{value} is not iterable.', value=value)
21
22        results = []
23
24        for item in value:
25            for validator in self._validators:
26                item = validator.validate(item)  # noqa: PLW2901
27
28            results.append(item)
29
30        return results

Base class for validator classes.

ForEach( validators: Validator | Iterable[Validator])
11    def __init__(self, validators: Validator | Iterable[Validator]) -> None:  # noqa: D107
12        if isinstance(validators, Validator):
13            self._validators = [validators]
14        else:
15            self._validators = validators
@overrides(Validator)
def validate(self, value: Iterable[typing.Any]) -> list[typing.Any]:
17    @overrides(Validator)
18    def validate(self, value: Iterable[Any]) -> list[Any]:  # noqa: D102
19        if not isinstance(value, collections.abc.Iterable):
20            self.raise_exception(msg=f'{value} is not iterable.', value=value)
21
22        results = []
23
24        for item in value:
25            for validator in self._validators:
26                item = validator.validate(item)  # noqa: PLW2901
27
28            results.append(item)
29
30        return results

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class GenericMixin:
  5class GenericMixin:
  6    """
  7    A mixin that provides easy access to given type variables.
  8
  9    Example:
 10        >>> from typing import Generic, TypeVar
 11        >>> T = TypeVar('T')
 12        >>> U = TypeVar('U')
 13        >>> class Foo(Generic[T, U], GenericMixin):
 14        ...     values: list[T]
 15        ...     value: U
 16        >>> f = Foo[str, int]()
 17        >>> f.type_vars
 18        {~T: <class 'str'>, ~U: <class 'int'>}
 19    """
 20
 21    @property
 22    def type_vars(self) -> dict[TypeVar, type]:
 23        """
 24        Returns the mapping of type variables to types.
 25
 26        DO NOT call this inside __init__()!
 27
 28        Example:
 29            >>> from typing import Generic, TypeVar
 30            >>> T = TypeVar('T')
 31            >>> U = TypeVar('U')
 32            >>> class Foo(Generic[T, U], GenericMixin):
 33            ...     values: list[T]
 34            ...     value: U
 35            >>> f = Foo[str, int]()
 36            >>> f.type_vars
 37            {~T: <class 'str'>, ~U: <class 'int'>}
 38        """
 39
 40        return self._get_resolved_typevars()
 41
 42    def _get_resolved_typevars(self) -> dict[TypeVar, type]:  # noqa: C901
 43        """
 44        Returns the resolved type vars.
 45
 46        Do not call this inside the __init__() method, because at that point the relevant information are not present.
 47        See also https://github.com/python/cpython/issues/90899'
 48        """
 49
 50        mapping: dict[TypeVar, type] = {}
 51        class_name = type(self).__name__
 52
 53        if not hasattr(self, '__orig_bases__'):
 54            raise AssertionError(
 55                f'{class_name} is not a generic class. To make it generic, declare it like: '
 56                f'class {class_name}(Generic[T], GenericMixin):...',
 57            )
 58
 59        def collect(base: type, substitutions: dict[TypeVar, type]) -> None:
 60            """Recursively collect type var mappings from a generic base."""
 61            origin = get_origin(base) or base
 62            args = get_args(base)
 63
 64            params = getattr(origin, '__parameters__', ())
 65            # copy substitutions so each recursion has its own view
 66            resolved = substitutions.copy()
 67
 68            for param, arg in zip(params, args, strict=False):
 69                if isinstance(arg, TypeVar):
 70                    processed_arg = substitutions.get(arg, arg)
 71                else:
 72                    processed_arg = arg
 73
 74                mapping[param] = processed_arg
 75                resolved[param] = processed_arg
 76
 77            # Recurse into base classes, applying current substitutions
 78            for super_base in getattr(origin, '__orig_bases__', []):
 79                super_origin = get_origin(super_base) or super_base
 80                super_args = get_args(super_base)
 81
 82                if super_args:
 83                    # Substitute any TypeVars in the super_base's args using resolved
 84                    substituted_args = tuple(
 85                        resolved.get(a, a) if isinstance(a, TypeVar) else a
 86                        for a in super_args
 87                    )
 88                    # Build a new parametrized base so get_args() inside collect sees substituted_args
 89                    try:
 90                        substituted_base = super_origin[substituted_args]  # type: ignore[index]
 91                    except TypeError:
 92                        # Some origins won't accept subscription; fall back to passing the origin and trusting resolved
 93                        substituted_base = super_base
 94                    collect(base=substituted_base, substitutions=resolved)
 95                else:
 96                    collect(base=super_base, substitutions=resolved)
 97
 98        # Start from __orig_class__ if present, else walk the declared MRO bases
 99        cls = getattr(self, '__orig_class__', None)
100        if cls is not None:
101            collect(base=cls, substitutions={})
102        else:
103            # Walk the full MRO to catch indirect generic ancestors
104            for c in self.__class__.__mro__:
105                for base in getattr(c, '__orig_bases__', []):
106                    collect(base=base, substitutions=mapping)
107
108        # Ensure no unresolved TypeVars remain
109        all_params = set()
110        for c in self.__class__.__mro__:
111            all_params.update(getattr(c, '__parameters__', ()))
112
113        unresolved = {p for p in all_params if p not in mapping or isinstance(mapping[p], TypeVar)}
114        if unresolved:
115            raise AssertionError(
116                f'You need to instantiate this class with type parameters! Example: {class_name}[int]()\n'
117                f'Also make sure that you do not call this in the __init__() method of your class!\n'
118                f'Unresolved type variables: {unresolved}\n'
119                f'See also https://github.com/python/cpython/issues/90899',
120            )
121
122        return mapping

A mixin that provides easy access to given type variables.

Example:

from typing import Generic, TypeVar T = TypeVar('T') U = TypeVar('U') class Foo(Generic[T, U], GenericMixin): ... values: list[T] ... value: U f = Foostr, int f.type_vars {~T: , ~U: }

type_vars: dict[typing.TypeVar, type]
21    @property
22    def type_vars(self) -> dict[TypeVar, type]:
23        """
24        Returns the mapping of type variables to types.
25
26        DO NOT call this inside __init__()!
27
28        Example:
29            >>> from typing import Generic, TypeVar
30            >>> T = TypeVar('T')
31            >>> U = TypeVar('U')
32            >>> class Foo(Generic[T, U], GenericMixin):
33            ...     values: list[T]
34            ...     value: U
35            >>> f = Foo[str, int]()
36            >>> f.type_vars
37            {~T: <class 'str'>, ~U: <class 'int'>}
38        """
39
40        return self._get_resolved_typevars()

Returns the mapping of type variables to types.

DO NOT call this inside __init__()!

Example:

from typing import Generic, TypeVar T = TypeVar('T') U = TypeVar('U') class Foo(Generic[T, U], GenericMixin): ... values: list[T] ... value: U f = Foostr, int f.type_vars {~T: , ~U: }

class InvalidHeader(pedantic.ParameterException):
70class InvalidHeader(ParameterException):
71    """Is raised if there is a validation error in a FlaskHeaderParameter."""

Is raised if there is a validation error in a FlaskHeaderParameter.

class IsEnum(pedantic.Validator):
 9class IsEnum(Validator): # noqa: D101
10    def __init__(self, enum: EnumMeta, convert: bool = True, to_upper_case: bool = True) -> None: # noqa: D107
11        self._enum = enum
12        self._convert = convert
13        self._to_upper_case = to_upper_case
14
15    @overrides(Validator)
16    def validate(self, value: Any) -> Any: # noqa: D102
17        try:
18            if isinstance(value, str) and self._to_upper_case:
19                value = value.upper()
20
21            if issubclass(self._enum, IntEnum):
22                enum_value = self._enum(int(value))
23            else:
24                enum_value = self._enum(value)
25        except (ValueError, TypeError):
26            return self.raise_exception(msg=f'Incorrect value {value} for enum {self._enum}.', value=value)
27
28        if self._convert:
29            return enum_value
30
31        return value

Base class for validator classes.

IsEnum( enum: enum.EnumType, convert: bool = True, to_upper_case: bool = True)
10    def __init__(self, enum: EnumMeta, convert: bool = True, to_upper_case: bool = True) -> None: # noqa: D107
11        self._enum = enum
12        self._convert = convert
13        self._to_upper_case = to_upper_case
@overrides(Validator)
def validate(self, value: Any) -> Any:
15    @overrides(Validator)
16    def validate(self, value: Any) -> Any: # noqa: D102
17        try:
18            if isinstance(value, str) and self._to_upper_case:
19                value = value.upper()
20
21            if issubclass(self._enum, IntEnum):
22                enum_value = self._enum(int(value))
23            else:
24                enum_value = self._enum(value)
25        except (ValueError, TypeError):
26            return self.raise_exception(msg=f'Incorrect value {value} for enum {self._enum}.', value=value)
27
28        if self._convert:
29            return enum_value
30
31        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class IsUuid(pedantic.Validator):
 8class IsUuid(Validator): # noqa: D101
 9    def __init__(self, convert: bool = False) -> None: # noqa: D107
10        self._convert = convert
11
12    @overrides(Validator)
13    def validate(self, value: str) -> str: # noqa: D102
14        try:
15            converted_value = UUID(str(value))
16        except ValueError:
17            return self.raise_exception(msg=f'{value} is not a valid UUID', value=value)
18
19        return converted_value if self._convert else value

Base class for validator classes.

IsUuid(convert: bool = False)
 9    def __init__(self, convert: bool = False) -> None: # noqa: D107
10        self._convert = convert
@overrides(Validator)
def validate(self, value: str) -> str:
12    @overrides(Validator)
13    def validate(self, value: str) -> str: # noqa: D102
14        try:
15            converted_value = UUID(str(value))
16        except ValueError:
17            return self.raise_exception(msg=f'{value} is not a valid UUID', value=value)
18
19        return converted_value if self._convert else value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class MatchPattern(pedantic.Validator):
 8class MatchPattern(Validator):  # noqa: D101
 9    def __init__(self, pattern: str) -> None:  # noqa: D107
10        self._pattern = re.compile(pattern=pattern)
11
12    @overrides(Validator)
13    def validate(self, value: str) -> str: # noqa: D102
14        if not self._pattern.search(string=str(value)):
15            self.raise_exception(msg=f'Value "{value}" does not match pattern {self._pattern.pattern}.', value=value)
16
17        return value

Base class for validator classes.

MatchPattern(pattern: str)
 9    def __init__(self, pattern: str) -> None:  # noqa: D107
10        self._pattern = re.compile(pattern=pattern)
@overrides(Validator)
def validate(self, value: str) -> str:
12    @overrides(Validator)
13    def validate(self, value: str) -> str: # noqa: D102
14        if not self._pattern.search(string=str(value)):
15            self.raise_exception(msg=f'Value "{value}" does not match pattern {self._pattern.pattern}.', value=value)
16
17        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class Max(pedantic.Validator):
 6class Max(Validator):  # noqa: D101
 7    def __init__(self, value: float, include_boundary: bool = True) -> None:
 8        """
 9        >>> Max(7, True).validate(7)
10        7
11        >>> Max(7, False).validate(7)  # doctest: +IGNORE_EXCEPTION_DETAIL
12        Traceback (most recent call last):
13        ValidatorException: ...
14        >>> Max(7, False).validate(6.999)
15        6.999
16        """
17        self._value = value
18        self._include_boundary = include_boundary
19
20    @overrides(Validator)
21    def validate(self, value: float) -> int | float:  # noqa: D102
22        if value > self._value and self._include_boundary:
23            self.raise_exception(msg=f'greater then allowed: {value} is not <= {self._value}', value=value)
24        elif value >= self._value and not self._include_boundary:
25            self.raise_exception(msg=f'greater then allowed: {value} is not < {self._value}', value=value)
26
27        return value

Base class for validator classes.

Max(value: float, include_boundary: bool = True)
 7    def __init__(self, value: float, include_boundary: bool = True) -> None:
 8        """
 9        >>> Max(7, True).validate(7)
10        7
11        >>> Max(7, False).validate(7)  # doctest: +IGNORE_EXCEPTION_DETAIL
12        Traceback (most recent call last):
13        ValidatorException: ...
14        >>> Max(7, False).validate(6.999)
15        6.999
16        """
17        self._value = value
18        self._include_boundary = include_boundary
>>> Max(7, True).validate(7)
7
>>> Max(7, False).validate(7)  # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ValidatorException: ...
>>> Max(7, False).validate(6.999)
6.999
@overrides(Validator)
def validate(self, value: float) -> int | float:
20    @overrides(Validator)
21    def validate(self, value: float) -> int | float:  # noqa: D102
22        if value > self._value and self._include_boundary:
23            self.raise_exception(msg=f'greater then allowed: {value} is not <= {self._value}', value=value)
24        elif value >= self._value and not self._include_boundary:
25            self.raise_exception(msg=f'greater then allowed: {value} is not < {self._value}', value=value)
26
27        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class MaxLength(pedantic.Validator):
10class MaxLength(Validator):  # noqa: D101
11    def __init__(self, length: int) -> None:  # noqa: D107
12        self._length = length
13
14    @overrides(Validator)
15    def validate(self, value: Sized) -> Any:  # noqa: D102
16        if not isinstance(value, collections.abc.Sized):
17            self.raise_exception(msg=f'{value} has no length.', value=value)
18
19        if len(value) > self._length:
20            self.raise_exception(msg=f'{value} is too long with length {len(value)}.', value=value)
21
22        return value

Base class for validator classes.

MaxLength(length: int)
11    def __init__(self, length: int) -> None:  # noqa: D107
12        self._length = length
@overrides(Validator)
def validate(self, value: Sized) -> Any:
14    @overrides(Validator)
15    def validate(self, value: Sized) -> Any:  # noqa: D102
16        if not isinstance(value, collections.abc.Sized):
17            self.raise_exception(msg=f'{value} has no length.', value=value)
18
19        if len(value) > self._length:
20            self.raise_exception(msg=f'{value} is too long with length {len(value)}.', value=value)
21
22        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class Min(pedantic.Validator):
 7class Min(Validator):  # noqa: D101
 8    def __init__(self, value: float, include_boundary: bool = True) -> None:
 9        """
10        >>> Min(7, True).validate(7)
11        7
12        >>> Min(7, False).validate(7)  # doctest: +IGNORE_EXCEPTION_DETAIL
13        Traceback (most recent call last):
14        ValidatorException: ...
15        >>> Min(7, False).validate(7.001)
16        7.001
17        """
18        self._value = value
19        self._include_boundary = include_boundary
20
21    @overrides(Validator)
22    def validate(self, value: float) -> int | float:  # noqa: D102
23        if value < self._value and self._include_boundary:
24            self.raise_exception(msg=f'smaller then allowed: {value} is not >= {self._value}', value=value)
25        elif value <= self._value and not self._include_boundary:
26            self.raise_exception(msg=f'smaller then allowed: {value} is not > {self._value}', value=value)
27
28        return value

Base class for validator classes.

Min(value: float, include_boundary: bool = True)
 8    def __init__(self, value: float, include_boundary: bool = True) -> None:
 9        """
10        >>> Min(7, True).validate(7)
11        7
12        >>> Min(7, False).validate(7)  # doctest: +IGNORE_EXCEPTION_DETAIL
13        Traceback (most recent call last):
14        ValidatorException: ...
15        >>> Min(7, False).validate(7.001)
16        7.001
17        """
18        self._value = value
19        self._include_boundary = include_boundary
>>> Min(7, True).validate(7)
7
>>> Min(7, False).validate(7)  # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ValidatorException: ...
>>> Min(7, False).validate(7.001)
7.001
@overrides(Validator)
def validate(self, value: float) -> int | float:
21    @overrides(Validator)
22    def validate(self, value: float) -> int | float:  # noqa: D102
23        if value < self._value and self._include_boundary:
24            self.raise_exception(msg=f'smaller then allowed: {value} is not >= {self._value}', value=value)
25        elif value <= self._value and not self._include_boundary:
26            self.raise_exception(msg=f'smaller then allowed: {value} is not > {self._value}', value=value)
27
28        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class MinLength(pedantic.Validator):
10class MinLength(Validator):  # noqa: D101
11    def __init__(self, length: int) -> None:  # noqa: D107
12        self._length = length
13
14    @overrides(Validator)
15    def validate(self, value: Sized) -> Any:  # noqa: D102
16        if not isinstance(value, collections.abc.Sized):
17            self.raise_exception(msg=f'{value} has no length.', value=value)
18
19        if len(value) < self._length:
20            self.raise_exception(msg=f'{value} is too short with length {len(value)}.', value=value)
21
22        return value

Base class for validator classes.

MinLength(length: int)
11    def __init__(self, length: int) -> None:  # noqa: D107
12        self._length = length
@overrides(Validator)
def validate(self, value: Sized) -> Any:
14    @overrides(Validator)
15    def validate(self, value: Sized) -> Any:  # noqa: D102
16        if not isinstance(value, collections.abc.Sized):
17            self.raise_exception(msg=f'{value} has no length.', value=value)
18
19        if len(value) < self._length:
20            self.raise_exception(msg=f'{value} is too short with length {len(value)}.', value=value)
21
22        return value

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

class NotEmpty(pedantic.Validator):
 9class NotEmpty(Validator):
10    """Validates that the given value is not empty."""
11
12    def __init__(self, strip: bool = True) -> None:
13        """If strip is True, the leading and trailing whitespace will be removed."""
14        self.strip = strip
15
16    @overrides(Validator)
17    def validate(self, value: Sequence) -> Sequence:
18        """
19        Throws a ValidationError if the sequence is empty.
20        If the sequence is a string, it removes all leading and trailing whitespace.
21        """
22
23        if isinstance(value, str):
24            if not value.strip():
25                self.raise_exception(msg='Got empty String which is invalid.', value=value)
26
27            return value.strip() if self.strip else value
28        if isinstance(value, collections.abc.Sequence):
29            if len(value) == 0:
30                raise self.raise_exception(msg='Got empty  which is invalid.', value=value)
31
32            return value
33
34        return self.raise_exception(msg=f'Got {type(value)} which is not a Sequence.', value=value)

Validates that the given value is not empty.

NotEmpty(strip: bool = True)
12    def __init__(self, strip: bool = True) -> None:
13        """If strip is True, the leading and trailing whitespace will be removed."""
14        self.strip = strip

If strip is True, the leading and trailing whitespace will be removed.

strip
@overrides(Validator)
def validate(self, value: Sequence) -> Sequence:
16    @overrides(Validator)
17    def validate(self, value: Sequence) -> Sequence:
18        """
19        Throws a ValidationError if the sequence is empty.
20        If the sequence is a string, it removes all leading and trailing whitespace.
21        """
22
23        if isinstance(value, str):
24            if not value.strip():
25                self.raise_exception(msg='Got empty String which is invalid.', value=value)
26
27            return value.strip() if self.strip else value
28        if isinstance(value, collections.abc.Sequence):
29            if len(value) == 0:
30                raise self.raise_exception(msg='Got empty  which is invalid.', value=value)
31
32            return value
33
34        return self.raise_exception(msg=f'Got {type(value)} which is not a Sequence.', value=value)

Throws a ValidationError if the sequence is empty. If the sequence is a string, it removes all leading and trailing whitespace.

class Parameter:
14class Parameter:  # noqa: D101
15    exception_type: type[ParameterException] = ParameterException
16
17    def __init__(  # noqa: D107
18        self,
19        name: str,
20        value_type: type[bool | int | float | str | dict | list] | None = None,
21        validators: Iterable[Validator] | None = None,
22        default: Any = NoValue,
23        required: bool = True,
24     ) -> None:
25        self.name = name
26        self.validators = validators or []
27        self.default_value = default
28        self.value_type = value_type
29        self.is_required = False if default != NoValue else required
30
31        if value_type not in [str, bool, int, float, dict, list, None]:
32            raise AssertionError('value_type needs to be one of these: str, bool, int, float, dict & list')
33
34    def validate(self, value: Any) -> Any:
35        """Apply all validators to the given value and collect all ValidationErrors."""
36
37        if value is None:
38            if self.is_required:
39                self.raise_exception(msg=f'Value for key {self.name} is required.')
40
41            return None
42
43        if self.value_type is not None:
44            try:
45                result_value = convert_value(value=value, target_type=self.value_type)
46            except ConversionError as ex:
47                return self.raise_exception(value=value, msg=ex.message)
48        else:
49            result_value = value
50
51        for validator in self.validators:
52            try:
53                result_value = validator.validate(result_value)
54            except ValidatorException as e:
55                raise self.exception_type.from_validator_exception(exception=e, parameter_name=self.name) from e
56
57        return result_value
58
59    def raise_exception(self, msg: str, value: Any = None, validator: Validator | None = None) -> NoReturn:  # noqa: D102
60        raise self.exception_type(
61            value=value,
62            parameter_name=self.name,
63            msg=msg,
64            validator_name=validator.name if validator else None,
65        )
66
67    def __str__(self) -> str:
68        return self.__class__.__name__ + ' name=' + self.name
Parameter( name: str, value_type: type[bool | int | float | str | dict | list] | None = None, validators: Iterable[Validator] | None = None, default: Any = <class 'pedantic.decorators.validate.parameters.abstract_parameter.NoValue'>, required: bool = True)
17    def __init__(  # noqa: D107
18        self,
19        name: str,
20        value_type: type[bool | int | float | str | dict | list] | None = None,
21        validators: Iterable[Validator] | None = None,
22        default: Any = NoValue,
23        required: bool = True,
24     ) -> None:
25        self.name = name
26        self.validators = validators or []
27        self.default_value = default
28        self.value_type = value_type
29        self.is_required = False if default != NoValue else required
30
31        if value_type not in [str, bool, int, float, dict, list, None]:
32            raise AssertionError('value_type needs to be one of these: str, bool, int, float, dict & list')
exception_type: type[ParameterException] = <class 'ParameterException'>
name
validators
default_value
value_type
is_required
def validate(self, value: Any) -> Any:
34    def validate(self, value: Any) -> Any:
35        """Apply all validators to the given value and collect all ValidationErrors."""
36
37        if value is None:
38            if self.is_required:
39                self.raise_exception(msg=f'Value for key {self.name} is required.')
40
41            return None
42
43        if self.value_type is not None:
44            try:
45                result_value = convert_value(value=value, target_type=self.value_type)
46            except ConversionError as ex:
47                return self.raise_exception(value=value, msg=ex.message)
48        else:
49            result_value = value
50
51        for validator in self.validators:
52            try:
53                result_value = validator.validate(result_value)
54            except ValidatorException as e:
55                raise self.exception_type.from_validator_exception(exception=e, parameter_name=self.name) from e
56
57        return result_value

Apply all validators to the given value and collect all ValidationErrors.

def raise_exception( self, msg: str, value: Any = None, validator: Validator | None = None) -> NoReturn:
59    def raise_exception(self, msg: str, value: Any = None, validator: Validator | None = None) -> NoReturn:  # noqa: D102
60        raise self.exception_type(
61            value=value,
62            parameter_name=self.name,
63            msg=msg,
64            validator_name=validator.name if validator else None,
65        )
class ParameterException(pedantic.ValidateException):
32class ParameterException(ValidateException):
33    """An exception that is raised inside a Parameter."""
34
35    def __init__( # noqa: D107
36        self,
37        msg: str,
38        parameter_name: str,
39        value: Any | None = None,
40        validator_name: str | None = None,
41    ) -> None:
42        super().__init__(msg=msg)
43        self.validator_name = validator_name
44        self.parameter_name = parameter_name
45        self.value = value
46
47    @classmethod
48    def from_validator_exception(cls, exception: ValidatorException, parameter_name: str = '') -> 'ParameterException':
49        """Creates a parameter exception from a validator exception."""
50        return cls(
51            value=exception.value,
52            msg=exception.message,
53            validator_name=exception.validator_name,
54            parameter_name=parameter_name or exception.parameter_name,
55        )
56
57    def __str__(self) -> str:
58        return str(self.to_dict)
59
60    @property
61    def to_dict(self) -> dict[str, str]:  # noqa: D102
62        return {
63            ExceptionDictKey.VALUE: str(self.value),
64            ExceptionDictKey.MESSAGE: self.message,
65            ExceptionDictKey.VALIDATOR: self.validator_name,
66            ExceptionDictKey.PARAMETER: self.parameter_name,
67        }

An exception that is raised inside a Parameter.

ParameterException( msg: str, parameter_name: str, value: Any | None = None, validator_name: str | None = None)
35    def __init__( # noqa: D107
36        self,
37        msg: str,
38        parameter_name: str,
39        value: Any | None = None,
40        validator_name: str | None = None,
41    ) -> None:
42        super().__init__(msg=msg)
43        self.validator_name = validator_name
44        self.parameter_name = parameter_name
45        self.value = value
validator_name
parameter_name
value
@classmethod
def from_validator_exception( cls, exception: ValidatorException, parameter_name: str = '') -> ParameterException:
47    @classmethod
48    def from_validator_exception(cls, exception: ValidatorException, parameter_name: str = '') -> 'ParameterException':
49        """Creates a parameter exception from a validator exception."""
50        return cls(
51            value=exception.value,
52            msg=exception.message,
53            validator_name=exception.validator_name,
54            parameter_name=parameter_name or exception.parameter_name,
55        )

Creates a parameter exception from a validator exception.

to_dict: dict[str, str]
60    @property
61    def to_dict(self) -> dict[str, str]:  # noqa: D102
62        return {
63            ExceptionDictKey.VALUE: str(self.value),
64            ExceptionDictKey.MESSAGE: self.message,
65            ExceptionDictKey.VALIDATOR: self.validator_name,
66            ExceptionDictKey.PARAMETER: self.parameter_name,
67        }
class ReturnAs(enum.Enum):
21class ReturnAs(Enum):  # noqa: D101
22    ARGS = 'ARGS'
23    KWARGS_WITH_NONE = 'KWARGS_WITH_NONE'
24    KWARGS_WITHOUT_NONE = 'KWARGS_WITHOUT_NONE'
ARGS = <ReturnAs.ARGS: 'ARGS'>
KWARGS_WITH_NONE = <ReturnAs.KWARGS_WITH_NONE: 'KWARGS_WITH_NONE'>
KWARGS_WITHOUT_NONE = <ReturnAs.KWARGS_WITHOUT_NONE: 'KWARGS_WITHOUT_NONE'>
class TooManyArguments(pedantic.ValidateException):
74class TooManyArguments(ValidateException):
75    """Is raised if the function got more arguments than expected."""

Is raised if the function got more arguments than expected.

class ValidateException(builtins.Exception):
12class ValidateException(Exception):  # noqa: N818
13    """The base class for all exception thrown by the validate decorator."""
14
15    def __init__(self, msg: str) -> None: # noqa: D107
16        self.message = msg

The base class for all exception thrown by the validate decorator.

ValidateException(msg: str)
15    def __init__(self, msg: str) -> None: # noqa: D107
16        self.message = msg
message
class Validator(abc.ABC):
 8class Validator(ABC):
 9    """Base class for validator classes."""
10
11    @abstractmethod
12    def validate(self, value: Any) -> Any:
13        """
14        Validates and convert the value.
15        Raises an [ValidatorException] in case of an invalid value.
16        To raise this you can simply call self.raise_exception().
17        """
18
19    def validate_param(self, value: Any, parameter_name: str) -> Any:
20        """
21        Validates and converts the value, just like [validate()].
22        The difference is that a parameter_name is included in the exception, if an exception is raised.
23        """
24
25        try:
26            return self.validate(value=value)
27        except ValidatorException as ex:
28            ex.parameter_name = parameter_name
29            raise
30
31    def raise_exception(self, value: Any, msg: str) -> NoReturn:  # noqa: D102
32        raise ValidatorException(value=value, validator_name=self.name, msg=msg)
33
34    @property
35    def name(self) -> str: # noqa: D102
36        return self.__class__.__name__

Base class for validator classes.

@abstractmethod
def validate(self, value: Any) -> Any:
11    @abstractmethod
12    def validate(self, value: Any) -> Any:
13        """
14        Validates and convert the value.
15        Raises an [ValidatorException] in case of an invalid value.
16        To raise this you can simply call self.raise_exception().
17        """

Validates and convert the value. Raises an [ValidatorException] in case of an invalid value. To raise this you can simply call self.raise_exception().

def validate_param(self, value: Any, parameter_name: str) -> Any:
19    def validate_param(self, value: Any, parameter_name: str) -> Any:
20        """
21        Validates and converts the value, just like [validate()].
22        The difference is that a parameter_name is included in the exception, if an exception is raised.
23        """
24
25        try:
26            return self.validate(value=value)
27        except ValidatorException as ex:
28            ex.parameter_name = parameter_name
29            raise

Validates and converts the value, just like [validate()]. The difference is that a parameter_name is included in the exception, if an exception is raised.

def raise_exception(self, value: Any, msg: str) -> NoReturn:
31    def raise_exception(self, value: Any, msg: str) -> NoReturn:  # noqa: D102
32        raise ValidatorException(value=value, validator_name=self.name, msg=msg)
name: str
34    @property
35    def name(self) -> str: # noqa: D102
36        return self.__class__.__name__
class ValidatorException(pedantic.ValidateException):
19class ValidatorException(ValidateException):
20    """An exception that is raised inside the validate() function of a Validator."""
21
22    def __init__(self, msg: str, validator_name: str, value: Any, parameter_name: str = '') -> None: # noqa: D107
23        super().__init__(msg=msg)
24        self.validator_name = validator_name
25        self.value = value
26        self.parameter_name = parameter_name
27
28    def __str__(self) -> str:
29        return f'{self.validator_name}: {self.message} Value: {self.value}'

An exception that is raised inside the validate() function of a Validator.

ValidatorException(msg: str, validator_name: str, value: Any, parameter_name: str = '')
22    def __init__(self, msg: str, validator_name: str, value: Any, parameter_name: str = '') -> None: # noqa: D107
23        super().__init__(msg=msg)
24        self.validator_name = validator_name
25        self.value = value
26        self.parameter_name = parameter_name
validator_name
value
parameter_name
class WithDecoratedMethods(abc.ABC, pedantic.GenericMixin, typing.Generic[~DecoratorTypeVar]):
 52class WithDecoratedMethods(ABC, GenericMixin, Generic[DecoratorTypeVar]):
 53    """
 54    A mixin that is used to figure out which method is decorated with custom parameterized decorators.
 55
 56    Example:
 57        >>> class Decorators(DecoratorType):
 58        ...     FOO = '_foo'
 59        ...     BAR = '_bar'
 60        >>> foo = create_decorator(decorator_type=Decorators.FOO)
 61        >>> bar = create_decorator(decorator_type=Decorators.BAR)
 62        >>> class MyClass(WithDecoratedMethods[Decorators]):
 63        ...    @foo(42)
 64        ...    def m1(self) -> None:
 65        ...        print('bar')
 66        ...
 67        ...    @foo(value=43)
 68        ...    def m2(self) -> None:
 69        ...        print('bar')
 70        ...
 71        ...    @bar(value=44)
 72        ...    def m3(self) -> None:
 73        ...        print('bar')
 74        >>> instance = MyClass()
 75        >>> instance.get_decorated_functions()  # doctest: +SKIP
 76        {
 77            <Decorators.FOO: '_foo'>: {
 78                <bound method MyClass.m1 of <__main__.MyClass object at 0x7fea7a6e2610>>: 42,
 79                <bound method MyClass.m2 of <__main__.MyClass object at 0x7fea7a6e2610>>: 43,
 80            },
 81            <Decorators.BAR: '_bar'>: {
 82                <bound method MyClass.m3 of <__main__.MyClass object at 0x7fea7a6e2610>>: 44,
 83            }
 84        }
 85    """
 86
 87    def get_decorated_functions(self) -> dict[DecoratorTypeVar, dict[C, T]]:
 88        """Returns a mapping of all functions that are decorated by the given decorator type."""
 89
 90        decorator_types: DecoratorTypeVar = self.type_vars[DecoratorTypeVar]  # type: ignore[assignment]
 91        decorated_functions = {t: {} for t in decorator_types}
 92
 93        for attribute_name in dir(self):
 94            if attribute_name.startswith('__') or attribute_name in ['type_var', 'type_vars']:
 95                continue
 96
 97            try:
 98                attribute = getattr(self, attribute_name)
 99            except BaseException:  # noqa: BLE001
100                continue  # ignore bad attributes
101
102            for decorator_type in decorator_types:
103                if hasattr(attribute, decorator_type):
104                    decorated_functions[decorator_type][attribute] = getattr(attribute, decorator_type)
105
106        return decorated_functions

A mixin that is used to figure out which method is decorated with custom parameterized decorators.

Example:

class Decorators(DecoratorType): ... FOO = '_foo' ... BAR = '_bar' foo = create_decorator(decorator_type=Decorators.FOO) bar = create_decorator(decorator_type=Decorators.BAR) class MyClass(WithDecoratedMethods[Decorators]): ... @foo(42) ... def m1(self) -> None: ... print('bar') ... ... @foo(value=43) ... def m2(self) -> None: ... print('bar') ... ... @bar(value=44) ... def m3(self) -> None: ... print('bar') instance = MyClass() instance.get_decorated_functions() # doctest: +SKIP { : { >: 42, >: 43, }, : { >: 44, } }

def get_decorated_functions(self) -> dict[~DecoratorTypeVar, dict[~C, ~T]]:
 87    def get_decorated_functions(self) -> dict[DecoratorTypeVar, dict[C, T]]:
 88        """Returns a mapping of all functions that are decorated by the given decorator type."""
 89
 90        decorator_types: DecoratorTypeVar = self.type_vars[DecoratorTypeVar]  # type: ignore[assignment]
 91        decorated_functions = {t: {} for t in decorator_types}
 92
 93        for attribute_name in dir(self):
 94            if attribute_name.startswith('__') or attribute_name in ['type_var', 'type_vars']:
 95                continue
 96
 97            try:
 98                attribute = getattr(self, attribute_name)
 99            except BaseException:  # noqa: BLE001
100                continue  # ignore bad attributes
101
102            for decorator_type in decorator_types:
103                if hasattr(attribute, decorator_type):
104                    decorated_functions[decorator_type][attribute] = getattr(attribute, decorator_type)
105
106        return decorated_functions

Returns a mapping of all functions that are decorated by the given decorator type.

def assert_value_matches_type( value: Any, type_: Any, err: str, type_vars: Dict[TypeVar, Any], key: str | None = None, msg: str | None = None, context: Dict[str, Any] | None = None) -> None:
17def assert_value_matches_type(  # noqa: PLR0913
18    value: Any,
19    type_: Any,
20    err: str,
21    type_vars: Dict[TypeVar_, Any],
22    key: str | None = None,
23    msg: str | None = None,
24    context: Dict[str, Any] | None = None,
25) -> None:
26    """Checks that the given value matches the given type."""
27
28    if not _check_type(value=value, type_=type_, err=err, type_vars=type_vars, context=context):
29        t = type(value)
30        value = f'{key}={value}' if key is not None else str(value)
31
32        if not msg:
33            msg = f'{err}Type hint is incorrect: Argument {value} of type {t} does not match expected type {type_}.'
34
35        raise PedanticTypeCheckException(msg)

Checks that the given value matches the given type.

async def calculate_in_subprocess( func: Callable[~P, ~T | Awaitable[~T]], *args: P.args, **kwargs: P.kwargs) -> ~T:
 54async def calculate_in_subprocess(func: Callable[P, T | Awaitable[T]], *args: P.args, **kwargs: P.kwargs) -> T:
 55    """
 56    Calculates the result of a synchronous function in subprocess without blocking the current thread.
 57
 58    Arguments:
 59        func: The function that will be called in a subprocess.
 60        args: Positional arguments that will be passed to the function.
 61        kwargs: Keyword arguments that will be passed to the function.
 62
 63    Returns:
 64         The calculated result of the function "func".
 65
 66    Raises:
 67        Any Exception that is raised inside [func].
 68
 69    Further reading: https://medium.com/devopss-hole/python-multiprocessing-pickle-issue-e2d35ccf96a9
 70
 71    Example:
 72    >>> import time
 73    >>> import asyncio
 74    >>> def f(value: int) -> int:
 75    ...     time.sleep(0.1)  # a long taking synchronous blocking calculation
 76    ...     return 2 * value
 77    >>> asyncio.run(calculate_in_subprocess(func=f, value=42))
 78    84
 79    """
 80
 81    if Pipe is None:
 82        raise ImportError('You need to install the multiprocess package to use this: pip install multiprocess')
 83
 84    rx, tx = Pipe(duplex=False)  # receiver & transmitter ; Pipe is one-way only
 85    process = Process(target=_inner, args=(tx, func, *args), kwargs=kwargs)
 86    process.start()
 87
 88    event = asyncio.Event()
 89    loop = asyncio.get_event_loop()
 90    loop.add_reader(fd=rx.fileno(), callback=event.set)
 91
 92    if not rx.poll():  # do not use process.is_alive() as condition here
 93        await event.wait()
 94
 95    loop.remove_reader(fd=rx.fileno())
 96    event.clear()
 97
 98    result = rx.recv()
 99    process.join()  # this blocks synchronously! make sure that process is terminated before you call join()
100    rx.close()
101    tx.close()
102
103    if isinstance(result, SubprocessError):
104        raise result.exception
105
106    return result

Calculates the result of a synchronous function in subprocess without blocking the current thread.

Arguments: func: The function that will be called in a subprocess. args: Positional arguments that will be passed to the function. kwargs: Keyword arguments that will be passed to the function.

Returns: The calculated result of the function "func".

Raises: Any Exception that is raised inside [func].

Further reading: https://medium.com/devopss-hole/python-multiprocessing-pickle-issue-e2d35ccf96a9

Example:

>>> import time
>>> import asyncio
>>> def f(value: int) -> int:
...     time.sleep(0.1)  # a long taking synchronous blocking calculation
...     return 2 * value
>>> asyncio.run(calculate_in_subprocess(func=f, value=42))
84
def create_decorator( decorator_type: DecoratorType, transformation: Callable[[~C, DecoratorType, ~T], ~C] | None = None) -> Callable[[~T], Callable[[~C], ~C]]:
28def create_decorator(
29    decorator_type: DecoratorType,
30    transformation: Callable[[C, DecoratorType, T], C] | None = None,
31) -> Callable[[T], Callable[[C], C]]:
32    """
33    Creates a new decorator that is parametrized with one argument of an arbitrary type.
34
35    You can also pass an arbitrary [transformation] to add custom behavior to the decorator.
36    """
37
38    def decorator(value: T) -> Callable[[C], C]:
39        def fun(f: C) -> C:
40            setattr(f, decorator_type, value)
41
42            if transformation is None:
43                return f
44
45            return transformation(f, decorator_type, value)
46
47        return fun  # we do not need functools.wraps, because we return the original function here
48
49    return decorator

Creates a new decorator that is parametrized with one argument of an arbitrary type.

You can also pass an arbitrary [transformation] to add custom behavior to the decorator.

def decorate_class( cls: ~T, decorate_callable: Callable[[Callable, ...], Callable], *args: Any, allowed_dunder_methods: list[str] | None = None) -> ~T:
17def decorate_class(
18    cls: T,
19    decorate_callable: Callable[[Callable, ...], Callable],
20    *args: Any,
21    allowed_dunder_methods: list[str] | None = None,
22) -> T:
23    """Applies a decorator with optional arguments to each method of a class."""
24
25    if allowed_dunder_methods is None:
26        allowed_dunder_methods = []
27
28    for attr_name, attr_value in vars(cls).items():
29        if attr_name.startswith('__') and attr_name.endswith('__') and attr_name not in allowed_dunder_methods:
30            continue
31
32        if inspect.isfunction(attr_value) or inspect.ismethod(attr_value):
33            setattr(cls, attr_name, decorate_callable(attr_value, *args))
34        elif isinstance(attr_value, staticmethod):
35            wrapped = decorate_callable(attr_value.__func__, *args)
36            setattr(cls, attr_name, staticmethod(wrapped))
37        elif isinstance(attr_value, classmethod):
38            wrapped = decorate_callable(attr_value.__func__, *args)
39            setattr(cls, attr_name, classmethod(wrapped))
40        elif isinstance(attr_value, property):
41            fget = decorate_callable(attr_value.fget, *args) if attr_value.fget is not None else None
42            fset = decorate_callable(attr_value.fset, *args) if attr_value.fset is not None else None
43            fdel = decorate_callable(attr_value.fdel, *args) if attr_value.fdel is not None else None
44
45            setattr(
46                cls,
47                attr_name,
48                property(
49                    fget=fget,
50                    fset=fset,
51                    fdel=fdel,
52                    doc=attr_value.__doc__,
53                ),
54            )
55
56    return cls

Applies a decorator with optional arguments to each method of a class.

def deprecated(func: Callable[~P, ~R] | type | None = None, *, message: str = '') -> Any:
34def deprecated(
35    func: Callable[P, R] | type | None = None,
36    *,
37    message: str = '',
38) -> Any:
39    """
40    Use this decorator to mark a function as deprecated. It will raise a warning when the function is called.
41    You can specify an optional reason or message to display with the warning.
42
43    If you use Python 3.13 or newer, consider using warnings.deprecated instead from the standard library.
44
45    Example:
46    >>> @deprecated
47    ... def my_function(a, b, c):
48    ...     pass
49    >>> my_function(5, 4, 3)  # doctest: +SKIP
50    >>> @deprecated(message='Will be removed soon. Please use my_function_new_instead.')
51    ... def my_function(a, b, c):
52    ...     pass
53    >>> my_function(5, 4, 3)  # doctest: +SKIP
54    """
55
56    def decorator(fun: Callable[P, R]) -> Callable[P, R]:
57        @wraps(fun)
58        def wrapper(*args: Any, **kwargs: Any) -> R:
59            msg = f'Call to deprecated function {fun.__qualname__}.'
60
61            if message:
62                msg += f'\nReason: {message}'
63
64            warnings.warn(message=msg, category=DeprecationWarning, stacklevel=2)
65            return fun(*args, **kwargs)
66        return wrapper
67    return decorator if func is None else decorator(func)

Use this decorator to mark a function as deprecated. It will raise a warning when the function is called. You can specify an optional reason or message to display with the warning.

If you use Python 3.13 or newer, consider using warnings.deprecated instead from the standard library.

Example:

>>> @deprecated
... def my_function(a, b, c):
...     pass
>>> my_function(5, 4, 3)  # doctest: +SKIP
>>> @deprecated(message='Will be removed soon. Please use my_function_new_instead.')
... def my_function(a, b, c):
...     pass
>>> my_function(5, 4, 3)  # doctest: +SKIP
@dataclass_transform(eq_default=True, frozen=True, order_default=False, kw_only_default=True)
def frozen_dataclass( cls: type[~T] | None = None, type_safe: bool = False, order: bool = False, kw_only: bool = True, slots: bool = False) -> type[pedantic.decorators.frozen_dataclass.FrozenDataclass] | type[~T] | Callable[[type[~T]], type[~T]]:
 66@dataclass_transform(eq_default=True, frozen=True, order_default=False, kw_only_default=True)
 67def frozen_dataclass(  # noqa: C901
 68    cls: type[T] | None = None,
 69    type_safe: bool = False,
 70    order: bool = False,
 71    kw_only: bool = True,
 72    slots: bool = False,
 73) -> type[FrozenDataclass] | type[T] | Callable[[type[T]], type[T]]:
 74    """
 75    Makes the decorated class immutable and a dataclass by adding the [@dataclass(frozen=True)]
 76    decorator. Also adds useful copy_with() and validate_types() instance methods to this class (see below).
 77
 78    If [type_safe] is True, a type check is performed for each field after the __post_init__ method was called,
 79    which itself is directly called after the __init__ constructor.
 80    Note that this might have a negative impact on performance.
 81    It's recommended to use this for debugging and testing only.
 82
 83    In a nutshell, the following methods will be added to the decorated class automatically:
 84    - __init__() gives you a simple constructor like "Foo(a=6, b='hi', c=True)"
 85    - __eq__() lets you compare objects easily with "a == b"
 86    - __hash__() is also needed for instance comparison
 87    - __repr__() gives you a nice output when you call "print(foo)"
 88    - copy_with() allows you to quickly create new similar frozen instances. Use this instead of setters.
 89    - deep_copy_with() allows you to create deep copies and modify them.
 90    - validate_types() allows you to validate the types of the dataclass.
 91                       This is called automatically when [type_safe] is True.
 92
 93    If the [order] parameter is True (default is False), the following comparison methods
 94    will be added additionally:
 95    - __lt__() lets you compare instance like "a < b"
 96    - __le__() lets you compare instance like "a <= b"
 97    - __gt__() lets you compare instance like "a > b"
 98    - __ge__() lets you compare instance like "a >= b"
 99
100    These compare the class as if it were a tuple of its fields, in order.
101    Both instances in the comparison must be of the identical type.
102
103    Example:
104    >>> @frozen_dataclass
105    ... class Foo:
106    ...     a: int
107    ...     b: str
108    ...     c: bool
109    >>> foo = Foo(a=6, b='hi', c=True)
110    >>> print(foo)
111    Foo(a=6, b='hi', c=True)
112    >>> print(foo.copy_with())
113    Foo(a=6, b='hi', c=True)
114    >>> print(foo.copy_with(a=42))
115    Foo(a=42, b='hi', c=True)
116    >>> print(foo.copy_with(b='Hello'))
117    Foo(a=6, b='Hello', c=True)
118    >>> print(foo.copy_with(c=False))
119    Foo(a=6, b='hi', c=False)
120    >>> print(foo.copy_with(a=676676, b='new', c=False))
121    Foo(a=676676, b='new', c=False)
122
123    If you want to get rid of those annoying "Unresolved attribute reference 'copy_with' for class 'Foo'" warnings
124    you can extend your models from FrozenDataclass:
125    Example:
126    >>> @frozen_dataclass
127    ... class Bar(FrozenDataclass):
128    ...     a: int
129    ...     b: str
130    ...     c: bool
131    >>> bar = Bar(a=6, b='hi', c=True)
132    >>> print(bar)
133    Bar(a=6, b='hi', c=True)
134    >>> print(bar.copy_with())
135    Bar(a=6, b='hi', c=True)
136    """
137
138    def decorator(cls_: type[T]) -> type[T]:
139        args = {'frozen': True, 'order': order, 'kw_only': kw_only, 'slots': slots}
140
141        if type_safe:
142            old_post_init = getattr(cls_, '__post_init__', lambda _: None)
143
144            def new_post_init(self: T) -> None:
145                old_post_init(self)
146                context = get_context(depth=3, increase_depth_if_name_matches=[
147                    copy_with.__name__,
148                    deep_copy_with.__name__,
149                ])
150                self.validate_types(_context=context)
151
152            cls_.__post_init__ = new_post_init  # must be done before applying dataclass()
153
154        new_class = dataclass(**args)(cls_)  # slots = True will create a new class!
155
156        def copy_with(self: T, **kwargs: Any) -> T:
157            """
158            Creates a new immutable instance that by copying all fields of this instance replaced by the new values.
159            Keep in mind that this is a shallow copy!
160            """
161
162            return replace(self, **kwargs)
163
164        def deep_copy_with(self: T, **kwargs: Any) -> T:
165            """
166            Creates a new immutable instance that by deep copying all fields of
167            this instance replaced by the new values.
168            """
169
170            current_values = {field.name: deepcopy(getattr(self, field.name)) for field in fields(self)}
171            return new_class(**{**current_values, **kwargs})
172
173        def validate_types(self: T, *, _context: dict[str, type] | None = None) -> None:
174            """
175            Checks that all instance variables have the correct type.
176            Raises a [PedanticTypeCheckException] if at least one type is incorrect.
177            """
178
179            props = fields(new_class)
180
181            if _context is None:
182                # method was called by user
183                _context = get_context(depth=2)
184
185            _context = {
186                **_context,
187                **self.__init__.__globals__,
188                self.__class__.__name__: self.__class__,
189            }
190
191            for field in props:
192                assert_value_matches_type(
193                    value=getattr(self, field.name),
194                    type_=field.type,
195                    err=f'In dataclass "{cls_.__name__}" in field "{field.name}": ',
196                    type_vars={},
197                    context=_context,
198                )
199
200        methods_to_add = [copy_with, deep_copy_with, validate_types]
201
202        for method in methods_to_add:
203            setattr(new_class, method.__name__, method)
204
205        return new_class
206
207    if cls is None:
208        return decorator
209
210    return decorator(cls_=cls)

Makes the decorated class immutable and a dataclass by adding the [@dataclass(frozen=True)] decorator. Also adds useful copy_with() and validate_types() instance methods to this class (see below).

If [type_safe] is True, a type check is performed for each field after the __post_init__ method was called, which itself is directly called after the __init__ constructor. Note that this might have a negative impact on performance. It's recommended to use this for debugging and testing only.

In a nutshell, the following methods will be added to the decorated class automatically:

  • __init__() gives you a simple constructor like "Foo(a=6, b='hi', c=True)"
  • __eq__() lets you compare objects easily with "a == b"
  • __hash__() is also needed for instance comparison
  • __repr__() gives you a nice output when you call "print(foo)"
  • copy_with() allows you to quickly create new similar frozen instances. Use this instead of setters.
  • deep_copy_with() allows you to create deep copies and modify them.
  • validate_types() allows you to validate the types of the dataclass. This is called automatically when [type_safe] is True.

If the [order] parameter is True (default is False), the following comparison methods will be added additionally:

  • __lt__() lets you compare instance like "a < b"
  • __le__() lets you compare instance like "a <= b"
  • __gt__() lets you compare instance like "a > b"
  • __ge__() lets you compare instance like "a >= b"

These compare the class as if it were a tuple of its fields, in order. Both instances in the comparison must be of the identical type.

Example:

>>> @frozen_dataclass
... class Foo:
...     a: int
...     b: str
...     c: bool
>>> foo = Foo(a=6, b='hi', c=True)
>>> print(foo)
Foo(a=6, b='hi', c=True)
>>> print(foo.copy_with())
Foo(a=6, b='hi', c=True)
>>> print(foo.copy_with(a=42))
Foo(a=42, b='hi', c=True)
>>> print(foo.copy_with(b='Hello'))
Foo(a=6, b='Hello', c=True)
>>> print(foo.copy_with(c=False))
Foo(a=6, b='hi', c=False)
>>> print(foo.copy_with(a=676676, b='new', c=False))
Foo(a=676676, b='new', c=False)

If you want to get rid of those annoying "Unresolved attribute reference 'copy_with' for class 'Foo'" warnings you can extend your models from FrozenDataclass: Example:

>>> @frozen_dataclass
... class Bar(FrozenDataclass):
...     a: int
...     b: str
...     c: bool
>>> bar = Bar(a=6, b='hi', c=True)
>>> print(bar)
Bar(a=6, b='hi', c=True)
>>> print(bar.copy_with())
Bar(a=6, b='hi', c=True)
def frozen_type_safe_dataclass(cls: type[~T]) -> type[~T]:
42def frozen_type_safe_dataclass(cls: type[T]) -> type[T]:
43    """Shortcut for @frozen_dataclass(type_safe=True)"""
44
45    return frozen_dataclass(type_safe=True)(cls)

Shortcut for @frozen_dataclass(type_safe=True)

def in_subprocess(func: Callable[~P, ~T | Awaitable[~T]]) -> Callable[~P, Awaitable[~T]]:
27def in_subprocess(func: Callable[P, T | Awaitable[T]]) -> Callable[P, Awaitable[T]]:
28    """
29    Executes the decorated function in a subprocess and returns the return value of it.
30    Note that the decorated function will be replaced with an async function which returns
31    a coroutine that needs to be awaited.
32    This purpose of this is doing long-taking calculations without blocking the main thread
33    of your application synchronously. That ensures that other asyncio.Tasks can work without any problem
34    at the same time.
35
36    Example:
37    >>> import time
38    >>> import asyncio
39    >>> @in_subprocess
40    ... def f(value: int) -> int:
41    ...     time.sleep(0.1)  # a long taking synchronous blocking calculation
42    ...     return 2 * value
43    >>> asyncio.run(f(value=42))
44    84
45    """
46
47    @wraps(func)
48    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
49        return await calculate_in_subprocess(func, *args, **kwargs)
50
51    return wrapper

Executes the decorated function in a subprocess and returns the return value of it. Note that the decorated function will be replaced with an async function which returns a coroutine that needs to be awaited. This purpose of this is doing long-taking calculations without blocking the main thread of your application synchronously. That ensures that other asyncio.Tasks can work without any problem at the same time.

Example:

>>> import time
>>> import asyncio
>>> @in_subprocess
... def f(value: int) -> int:
...     time.sleep(0.1)  # a long taking synchronous blocking calculation
...     return 2 * value
>>> asyncio.run(f(value=42))
84
def overrides(base_class: type) -> ~F:
10def overrides(base_class: type) -> F:
11    """
12    Use this to mark methods which overrides methods of a parent class.
13
14    Raises:
15        PedanticOverrideException: if the decorated method is not a method in any parent class
16
17    Example:
18    >>> class Parent:
19    ...     def my_instance_method(self): pass
20    >>> class Child(Parent):
21    ...     @overrides(Parent)
22    ...     def my_instance_method(self): pass
23    """
24
25    def decorator(func: F) -> F:
26        func_name = func.__name__
27
28        if func_name not in dir(base_class):
29            raise PedanticOverrideException(
30                f'In function {func.__qualname__}:\n '
31                f'Base class "{base_class.__name__}" does not have such a method "{func_name}".')
32        return func
33    return decorator

Use this to mark methods which overrides methods of a parent class.

Raises: PedanticOverrideException: if the decorated method is not a method in any parent class

Example:

>>> class Parent:
...     def my_instance_method(self): pass
>>> class Child(Parent):
...     @overrides(Parent)
...     def my_instance_method(self): pass
def pedantic( func: Callable[~P, ~R] | type | None = None, *, require_docstring: bool = False) -> Any:
 39def pedantic(
 40    func: Callable[P, R] | type | None = None,
 41    *,
 42    require_docstring: bool = False,
 43) -> Any:
 44    """
 45     A PedanticException is raised if one of the following happened:
 46     - The decorated function is called with positional arguments.
 47     - The function has no type annotation for their return type or one or more parameters do not have type
 48         annotations.
 49     - A type annotation is incorrect.
 50     - A type annotation misses type arguments, e.g. typing.List instead of typing.List[int].
 51     - The documented arguments do not match the argument list or their type annotations.
 52
 53     You can use this as a function or a class decorator.
 54
 55    Example:
 56    >>> @pedantic
 57    ... def my_function(a: int, b: float, c: str) -> bool:
 58    ...     return float(a) == b and str(b) == c
 59    >>> my_function(a=42.0, b=14.0, c='hi')
 60    Traceback (most recent call last):
 61    ...
 62    pedantic.exceptions.PedanticTypeCheckException: In function my_function:
 63    Type hint is incorrect: Argument a=42.0 of type <class 'float'> does not match expected type <class 'int'>.
 64    >>> my_function(a=42, b=None, c='hi')
 65    Traceback (most recent call last):
 66    ...
 67    pedantic.exceptions.PedanticTypeCheckException: In function my_function:
 68    Type hint is incorrect: Argument b=None of type <class 'NoneType'> does not match expected type <class 'float'>.
 69    >>> my_function(a=42, b=42, c='hi')
 70    Traceback (most recent call last):
 71    ...
 72    pedantic.exceptions.PedanticTypeCheckException: In function my_function:
 73    Type hint is incorrect: Argument b=42 of type <class 'int'> does not match expected type <class 'float'>.
 74    >>> my_function(5, 4.0, 'hi')
 75    Traceback (most recent call last):
 76    ...
 77    pedantic.exceptions.PedanticCallWithArgsException: In function my_function:
 78    Use kwargs when you call function my_function. Args: (5, 4.0, 'hi')
 79    """
 80
 81    def decorator(f: Callable[P, R] | type) -> Callable[P, R]:
 82        if inspect.isclass(f):
 83            add_type_var_attr_and_method_to_class(f)
 84            return decorate_class(
 85                cls=f,
 86                decorate_callable=_decorate_callable,
 87                allowed_dunder_methods=['__init__', '__contains__'],
 88            )
 89
 90        return _decorate_callable(f)
 91
 92    def _decorate_callable(f: Callable[P, R]) -> Callable[P, R]:
 93        decorated_func = DecoratedFunction(func=f)
 94
 95        if decorated_func.docstring is not None and (require_docstring or len(decorated_func.docstring.params)) > 0:
 96            check_docstring(decorated_func=decorated_func)
 97
 98        if decorated_func.is_coroutine:
 99            async def async_wrapper(*args: Any, **kwargs: Any) -> R:
100                call = FunctionCall(func=decorated_func, args=args, kwargs=kwargs, context=get_context(2))
101                call.assert_uses_kwargs()
102                return await call.async_check_types()
103
104            return async_wrapper
105
106        @wraps(f)
107        def sync_wrapper(*args: Any, **kwargs: Any) -> R:
108            call = FunctionCall(func=decorated_func, args=args, kwargs=kwargs, context=get_context(2))
109            call.assert_uses_kwargs()
110            return call.check_types()
111
112        return sync_wrapper
113
114    return decorator if func is None else decorator(f=func)

A PedanticException is raised if one of the following happened:

  • The decorated function is called with positional arguments.
  • The function has no type annotation for their return type or one or more parameters do not have type annotations.
  • A type annotation is incorrect.
  • A type annotation misses type arguments, e.g. typing.List instead of typing.List[int].
  • The documented arguments do not match the argument list or their type annotations.

You can use this as a function or a class decorator.

Example:

>>> @pedantic
... def my_function(a: int, b: float, c: str) -> bool:
...     return float(a) == b and str(b) == c
>>> my_function(a=42.0, b=14.0, c='hi')
Traceback (most recent call last):
...
pedantic.exceptions.PedanticTypeCheckException: In function my_function:
Type hint is incorrect: Argument a=42.0 of type <class 'float'> does not match expected type <class 'int'>.
>>> my_function(a=42, b=None, c='hi')
Traceback (most recent call last):
...
pedantic.exceptions.PedanticTypeCheckException: In function my_function:
Type hint is incorrect: Argument b=None of type <class 'NoneType'> does not match expected type <class 'float'>.
>>> my_function(a=42, b=42, c='hi')
Traceback (most recent call last):
...
pedantic.exceptions.PedanticTypeCheckException: In function my_function:
Type hint is incorrect: Argument b=42 of type <class 'int'> does not match expected type <class 'float'>.
>>> my_function(5, 4.0, 'hi')
Traceback (most recent call last):
...
pedantic.exceptions.PedanticCallWithArgsException: In function my_function:
Use kwargs when you call function my_function. Args: (5, 4.0, 'hi')
def resolve_forward_ref( type_: str, globals_: dict[str, Any] | None = None, context: dict | None = None) -> type:
 5def resolve_forward_ref(type_: str, globals_: dict[str, Any] | None = None, context: dict | None = None) -> type:  # noqa: F405
 6    """
 7    Resolve a type annotation that is a string.
 8
 9    Raises:
10        NameError: in case of [type_] cannot be resolved.
11    """
12
13    return eval(str(type_), globals_ or globals(), context or {})  # # noqa: S307

Resolve a type annotation that is a string.

Raises: NameError: in case of [type_] cannot be resolved.

def retry( *, attempts: int, exceptions: type[Exception] | tuple[type[Exception], ...] = <class 'Exception'>, sleep_time: datetime.timedelta = datetime.timedelta(0), logger: logging.Logger | None = None) -> Callable[[~C], ~C]:
15def retry(
16    *,
17    attempts: int,
18    exceptions: type[Exception] | tuple[type[Exception], ...] = Exception,
19    sleep_time: timedelta = timedelta(seconds=0),
20    logger: Logger | None = None,
21) -> Callable[[C], C]:
22    """
23    Retries the decorated function/method `attempts` times if the exceptions listed
24    in [exceptions] are thrown.
25
26    Args:
27        attempts: The number of times to repeat the wrapped function/method
28        exceptions: Lists of exceptions that trigger a retry attempt.
29        sleep_time: The time to wait between the retry attempts.
30        logger: The logger used for logging.
31
32    Example:
33        >>> @retry(attempts=3, exceptions=(ValueError, TypeError))
34        ... def foo():
35        ...     raise ValueError('Some error')
36        >>> foo()  # doctest: +SKIP
37    """
38
39    def decorator(func: C) -> C:
40        @wraps(func)
41        def wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
42            return retry_func(
43                func,
44                *args,
45                attempts=attempts,
46                exceptions=exceptions,
47                sleep_time=sleep_time,
48                logger=logger,
49                **kwargs,
50            )
51        return wrapper
52    return decorator

Retries the decorated function/method attempts times if the exceptions listed in [exceptions] are thrown.

Args: attempts: The number of times to repeat the wrapped function/method exceptions: Lists of exceptions that trigger a retry attempt. sleep_time: The time to wait between the retry attempts. logger: The logger used for logging.

Example:

@retry(attempts=3, exceptions=(ValueError, TypeError)) ... def foo(): ... raise ValueError('Some error') foo() # doctest: +SKIP

def run_doctest_of_single_function(f: Callable) -> None:
 6def run_doctest_of_single_function(f: Callable) -> None:
 7    """Useful for debugging a function with doctests."""
 8
 9    finder = doctest.DocTestFinder()
10    runner = doctest.DocTestRunner()
11
12    # Find doctests attached to the function
13    tests = finder.find(f)
14
15    # Run them
16    for test in tests:
17        runner.run(test)
18
19    # Fail the pytest test if any doctest failed
20    results = runner.summarize()
21
22    if results.failed > 0:
23        raise AssertionError(f'Failed tests: {results.failed}')

Useful for debugging a function with doctests.

def safe_async_contextmanager( f: Callable[..., AsyncIterator[~T]]) -> Callable[..., contextlib.AbstractAsyncContextManager[~T]]:
 68def safe_async_contextmanager(f: Callable[..., AsyncIterator[T]]) -> Callable[..., AbstractAsyncContextManager[T]]:
 69    """
 70    @safe_async_contextmanager decorator.
 71
 72         Typical usage:
 73
 74            @safe_async_contextmanager
 75            async def some_async_generator(<arguments>):
 76                <setup>
 77                yield <value>
 78                <cleanup>
 79
 80        equivalent to this:
 81
 82            @asynccontextmanager
 83            async def some_async_generator(<arguments>):
 84                <setup>
 85                try:
 86                    yield <value>
 87                finally:
 88                    <cleanup>
 89
 90        This makes this:
 91
 92            async with some_async_generator(<arguments>) as <variable>:
 93                <body>
 94
 95        equivalent to this:
 96
 97            <setup>
 98            try:
 99                <variable> = <value>
100                <body>
101            finally:
102                <cleanup>
103    """
104
105    if not isasyncgenfunction(f):
106        if not isgeneratorfunction(f):
107            raise AssertionError(f'{f.__name__} is not a generator.')
108
109        raise AssertionError(f'{f.__name__} is not an async generator. '
110                             f'So you need to use "safe_contextmanager" instead.')
111
112    @wraps(f)
113    async def wrapper(*args: Any, **kwargs: Any) -> Iterator[T]:
114        iterator = f(*args, **kwargs)
115
116        try:
117            yield await anext(iterator)
118        finally:
119            try:  # noqa: SIM105
120                await anext(iterator)
121            except StopAsyncIteration:
122                pass
123
124    return asynccontextmanager(wrapper)

@safe_async_contextmanager decorator.

 Typical usage:

    @safe_async_contextmanager
    async def some_async_generator(<arguments>):
        <setup>
        yield <value>
        <cleanup>

equivalent to this:

    @asynccontextmanager
    async def some_async_generator(<arguments>):
        <setup>
        try:
            yield <value>
        finally:
            <cleanup>

This makes this:

    async with some_async_generator(<arguments>) as <variable>:
        <body>

equivalent to this:

    <setup>
    try:
        <variable> = <value>
        <body>
    finally:
        <cleanup>
def safe_contextmanager( f: Callable[..., Iterator[~T]]) -> Callable[..., contextlib.AbstractContextManager[~T]]:
11def safe_contextmanager(f: Callable[..., Iterator[T]]) -> Callable[..., AbstractContextManager[T]]:
12    """
13    @safe_contextmanager decorator.
14
15    Typical usage:
16
17        @safe_contextmanager
18        def some_generator(<arguments>):
19            <setup>
20            yield <value>
21            <cleanup>
22
23    equivalent to this:
24
25        @contextmanager
26        def some_generator(<arguments>):
27            <setup>
28            try:
29                yield <value>
30            finally:
31                <cleanup>
32
33    This makes this:
34
35        with some_generator(<arguments>) as <variable>:
36            <body>
37
38    equivalent to this:
39
40        <setup>
41        try:
42            <variable> = <value>
43            <body>
44        finally:
45            <cleanup>
46    """
47
48    if isasyncgenfunction(f):
49        raise AssertionError(f'{f.__name__} is async. So you need to use "safe_async_contextmanager" instead.')
50    if not isgeneratorfunction(f):
51        raise AssertionError(f'{f.__name__} is not a generator.')
52
53    @wraps(f)
54    def wrapper(*args: Any, **kwargs: Any) -> Iterator[T]:
55        iterator = f(*args, **kwargs)
56
57        try:
58            yield next(iterator)
59        finally:
60            try:  # noqa: SIM105
61                next(iterator)
62            except StopIteration:
63                pass  # this is intended
64
65    return contextmanager(wrapper)

@safe_contextmanager decorator.

Typical usage:

@safe_contextmanager
def some_generator(<arguments>):
    <setup>
    yield <value>
    <cleanup>

equivalent to this:

@contextmanager
def some_generator(<arguments>):
    <setup>
    try:
        yield <value>
    finally:
        <cleanup>

This makes this:

with some_generator(<arguments>) as <variable>:
    <body>

equivalent to this:

<setup>
try:
    <variable> = <value>
    <body>
finally:
    <cleanup>
def trace( obj: Callable[~P, ~R] | type | None = None, *, log: Callable[[str], None] = <built-in function print>) -> Any:
37def trace(
38    obj: Callable[P, R] | type | None = None,
39    *,
40    log: Callable[[str], None] = print,
41) -> Any:
42    """
43    Decorate a function or class to trace calls, arguments, and return values.
44
45    Can be used as:
46
47        @trace
48        def foo(...): ...
49
50        @trace(log=print)
51        async def foo(...): ...
52
53        @trace
54        class Foo: ...
55
56        @trace(log=logger.info)
57        class Foo: ...
58
59    Supports:
60        - synchronous functions
61        - asynchronous functions
62        - classes
63        - instance methods
64        - static methods
65        - class methods
66
67    Args:
68        obj:
69            The decorated object when used as ``@trace``.
70
71        log:
72            Callable used to emit trace messages.
73
74    Returns:
75        The wrapped function, class, or decorator.
76    """
77
78    def decorator(obj_: Any) -> Any:
79        if inspect.isclass(obj_):
80            return decorate_class(obj_, _decorate_callable, log)
81
82        return _decorate_callable(obj_, log)
83
84    if obj is not None:
85        return decorator(obj)
86
87    return decorator

Decorate a function or class to trace calls, arguments, and return values.

Can be used as:

@trace
def foo(...): ...

@trace(log=print)
async def foo(...): ...

@trace
class Foo: ...

@trace(log=logger.info)
class Foo: ...

Supports: - synchronous functions - asynchronous functions - classes - instance methods - static methods - class methods

Args: obj: The decorated object when used as @trace.

log:
    Callable used to emit trace messages.

Returns: The wrapped function, class, or decorator.

def validate( *parameters: Parameter, return_as: ReturnAs = <ReturnAs.ARGS: 'ARGS'>, strict: bool = True, ignore_input: bool = False) -> Callable:
 27def validate(  # noqa: PLR0915, C901
 28    *parameters: Parameter,
 29    return_as: ReturnAs = ReturnAs.ARGS,
 30    strict: bool = True,
 31    ignore_input: bool = False,
 32) -> Callable:
 33    """
 34    Validates the values that are passed to the function by using the validators in the given parameters.
 35    The decorated function could also be async or an instance method as well as a normal function.
 36
 37    Args:
 38        parameters (multiple Parameter): The parameters that will be validated.
 39        return_as (ReturnAs): Pass the arguments as kwargs to the decorated function if ReturnAs.KWARGS.
 40            Positional arguments are used otherwise.
 41        strict (bool): If strict is true, you have to define a Parameter for each of the
 42            arguments the decorated function takes.
 43        ignore_input (bool): If True, all given arguments passed to this decorator are ignored.
 44            This can be useful if you use only ExternalParameters.
 45
 46    Returns:
 47        Callable: The decorated function.
 48    """
 49
 50    def validator(func: Callable) -> Callable:  # noqa: PLR0915, C901
 51        is_coroutine = inspect.iscoroutinefunction(func)
 52
 53        @wraps(func)
 54        def wrapper(*args: Any, **kwargs: Any) -> Any:
 55            result = _wrapper_content(*args, **kwargs)
 56
 57            if return_as == ReturnAs.ARGS:
 58                if 'self' in result:
 59                    return func(result.pop('self'), **result)
 60
 61                return func(*result.values())
 62
 63            if return_as == ReturnAs.KWARGS_WITHOUT_NONE:
 64                result = {k: v for k, v in result.items() if v is not None}
 65
 66            if 'self' in result:
 67                return func(result.pop('self'), **result)
 68
 69            return func(**result)
 70
 71        @wraps(func)
 72        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
 73            result = _wrapper_content(*args, **kwargs)
 74
 75            if return_as == ReturnAs.ARGS:
 76                if 'self' in result:
 77                    return await func(result.pop('self'), **result)
 78
 79                return await func(*result.values())
 80
 81            if return_as == ReturnAs.KWARGS_WITHOUT_NONE:
 82                result = {k: v for k, v in result.items() if v is not None}
 83
 84            if 'self' in result:
 85                return await func(result.pop('self'), **result)
 86
 87            return await func(**result)
 88
 89        def _wrapper_content(*args: Any, **kwargs: Any) -> dict[str, Any]:  # noqa: PLR0912, C901
 90            result = {}
 91            parameter_dict = {parameter.name: parameter for parameter in parameters}
 92            used_parameter_names: list[str] = []
 93            signature = inspect.signature(func)
 94
 95            if not ignore_input:
 96                for k, v in kwargs.items():
 97                    if k in parameter_dict:
 98                        parameter = parameter_dict[k]
 99                        result[k] = parameter.validate(value=v)
100                        used_parameter_names.append(parameter.name)
101                    elif strict:
102                        raise TooManyArguments(f'Got more arguments expected: No parameter found for argument {k}')
103                    else:
104                        result[k] = v
105
106                wants_args = '*args' in str(signature)
107                used_args = []
108
109                try:
110                    bound_args = signature.bind_partial(*args).arguments
111                except TypeError as ex:
112                    raise ValidateException(str(ex)) from ex
113
114                for k in bound_args:
115                    if k == 'args' and wants_args:
116                        for arg, parameter in zip(
117                            [a for a in args if a not in used_args],
118                            [p for p in parameters if p.name not in used_parameter_names],
119                            strict=True,
120                        ):
121                            result[parameter.name] = parameter.validate(arg)
122                            used_parameter_names.append(parameter.name)
123                    elif k in parameter_dict:
124                        parameter = parameter_dict[k]
125                        result[k] = parameter.validate(value=bound_args[k])
126                        used_parameter_names.append(parameter.name)
127                        used_args.append(bound_args[k])
128                    elif strict and k != 'self':
129                        raise TooManyArguments(f'Got more arguments expected: No parameter found for argument {k}')
130                    else:
131                        result[k] = bound_args[k]
132
133            unused_parameters = [parameter for parameter in parameters if parameter.name not in used_parameter_names]
134
135            for parameter in unused_parameters:
136                if isinstance(parameter, ExternalParameter) and parameter.has_value():
137                    v = parameter.load_value()
138                    result[parameter.name] = parameter.validate(value=v)
139                    continue
140
141                if parameter.is_required:
142                    return parameter.raise_exception(msg=f'Value for parameter {parameter.name} is required.')
143                if parameter.default_value == NoValue:
144                    if parameter.name in signature.parameters and \
145                            signature.parameters[parameter.name].default is not signature.empty:
146                        value = signature.parameters[parameter.name].default
147                    else:
148                        raise ValidateException(f'Got neither value nor default value for parameter {parameter.name}')
149                else:
150                    value = parameter.default_value
151
152                result[parameter.name] = value
153
154            # this is ugly, but I really want this behavior
155            if (
156                strict and IS_FLASK_INSTALLED and
157                all(isinstance(p, FlaskJsonParameter) for p in parameter_dict.values())
158                and request.is_json  # this check must come last
159            ):
160                unexpected_args = [k for k in request.json if k not in parameter_dict]
161
162                if unexpected_args:
163                    raise TooManyArguments(f'Got unexpected arguments: {unexpected_args}')
164
165            return result
166
167        if is_coroutine:
168            return async_wrapper
169        return wrapper
170    return validator

Validates the values that are passed to the function by using the validators in the given parameters. The decorated function could also be async or an instance method as well as a normal function.

Args: parameters (multiple Parameter): The parameters that will be validated. return_as (ReturnAs): Pass the arguments as kwargs to the decorated function if ReturnAs.KWARGS. Positional arguments are used otherwise. strict (bool): If strict is true, you have to define a Parameter for each of the arguments the decorated function takes. ignore_input (bool): If True, all given arguments passed to this decorator are ignored. This can be useful if you use only ExternalParameters.

Returns: Callable: The decorated function.

class FlaskFormParameter(pedantic.FlaskParameter):
38class FlaskFormParameter(FlaskParameter):  # noqa: D101
39    @overrides(FlaskParameter)
40    def get_dict(self) -> dict:  # noqa: D102
41        return request.form

The interface for all external parameters.

@overrides(FlaskParameter)
def get_dict(self) -> dict:
39    @overrides(FlaskParameter)
40    def get_dict(self) -> dict:  # noqa: D102
41        return request.form

Returns the actual values as a dictionary.

class FlaskGetParameter(pedantic.FlaskParameter):
51class FlaskGetParameter(FlaskParameter):  # noqa: D101
52    @overrides(FlaskParameter)
53    def get_dict(self) -> dict:  # noqa: D102
54        return request.args
55
56    @overrides(ExternalParameter)
57    def load_value(self) -> Any:  # noqa: D102
58        value = request.args.getlist(self.name)
59
60        if self.value_type is list:
61            return value
62
63        return value[0]

The interface for all external parameters.

@overrides(FlaskParameter)
def get_dict(self) -> dict:
52    @overrides(FlaskParameter)
53    def get_dict(self) -> dict:  # noqa: D102
54        return request.args

Returns the actual values as a dictionary.

@overrides(ExternalParameter)
def load_value(self) -> Any:
56    @overrides(ExternalParameter)
57    def load_value(self) -> Any:  # noqa: D102
58        value = request.args.getlist(self.name)
59
60        if self.value_type is list:
61            return value
62
63        return value[0]

Loads a value and returns it.

class FlaskHeaderParameter(pedantic.FlaskParameter):
66class FlaskHeaderParameter(FlaskParameter):  # noqa: D101
67    exception_type = InvalidHeader
68
69    @overrides(FlaskParameter)
70    def get_dict(self) -> dict:  # noqa: D102
71        return request.headers

The interface for all external parameters.

exception_type = <class 'InvalidHeader'>
@overrides(FlaskParameter)
def get_dict(self) -> dict:
69    @overrides(FlaskParameter)
70    def get_dict(self) -> dict:  # noqa: D102
71        return request.headers

Returns the actual values as a dictionary.

class FlaskJsonParameter(pedantic.FlaskParameter):
29class FlaskJsonParameter(FlaskParameter):  # noqa: D101
30    @overrides(FlaskParameter)
31    def get_dict(self) -> dict:  # noqa: D102
32        if not request.is_json:
33            return {}
34
35        return request.json

The interface for all external parameters.

@overrides(FlaskParameter)
def get_dict(self) -> dict:
30    @overrides(FlaskParameter)
31    def get_dict(self) -> dict:  # noqa: D102
32        if not request.is_json:
33            return {}
34
35        return request.json

Returns the actual values as a dictionary.

class FlaskParameter(pedantic.ExternalParameter, abc.ABC):
13class FlaskParameter(ExternalParameter, ABC):  # noqa: D101
14    @abstractmethod
15    def get_dict(self) -> dict[str, Any]:
16        """Returns the actual values as a dictionary."""
17
18    @overrides(ExternalParameter)
19    def has_value(self) -> bool:  # noqa: D102
20        dict_ = self.get_dict()
21        return dict_ is not None and self.name in dict_
22
23    @overrides(ExternalParameter)
24    def load_value(self) -> Any:  # noqa: D102
25        dict_ = self.get_dict()
26        return dict_[self.name]

The interface for all external parameters.

@abstractmethod
def get_dict(self) -> dict[str, typing.Any]:
14    @abstractmethod
15    def get_dict(self) -> dict[str, Any]:
16        """Returns the actual values as a dictionary."""

Returns the actual values as a dictionary.

@overrides(ExternalParameter)
def has_value(self) -> bool:
18    @overrides(ExternalParameter)
19    def has_value(self) -> bool:  # noqa: D102
20        dict_ = self.get_dict()
21        return dict_ is not None and self.name in dict_

Returns True if the value can be fetched.

@overrides(ExternalParameter)
def load_value(self) -> Any:
23    @overrides(ExternalParameter)
24    def load_value(self) -> Any:  # noqa: D102
25        dict_ = self.get_dict()
26        return dict_[self.name]

Loads a value and returns it.

class FlaskPathParameter(pedantic.Parameter):
44class FlaskPathParameter(Parameter):
45    """
46    This is a special case because Flask passes path parameter as kwargs to validate().
47    Therefore, this doesn't need to be an ExternalParameter.
48    """

This is a special case because Flask passes path parameter as kwargs to validate(). Therefore, this doesn't need to be an ExternalParameter.

class GenericFlaskDeserializer(pedantic.ExternalParameter):
 74class GenericFlaskDeserializer(ExternalParameter):
 75    """
 76    A JSON deserializer for classes which implements the [Deserializable] interface.
 77
 78    Further reading: https://github.com/LostInDarkMath/pedantic-python-decorators/issues/55
 79    """
 80
 81    def __init__(self, cls: type[Deserializable], catch_exception: bool = True, **kwargs: Any) -> None:  # noqa: D107
 82        super().__init__(**kwargs)
 83        self._cls = cls
 84        self._catch_exceptions = catch_exception
 85
 86    @overrides(ExternalParameter)
 87    def has_value(self) -> bool:  # noqa: D102
 88        return request.is_json
 89
 90    @overrides(ExternalParameter)
 91    def load_value(self) -> Any:  # noqa: D102
 92        try:
 93            return self._cls.from_json(request.json)
 94        except ValidatorException as ex:
 95            raise ParameterException.from_validator_exception(exception=ex, parameter_name='') from ex
 96        except Exception as ex:
 97            if self._catch_exceptions:
 98                self.raise_exception(msg=str(ex))
 99
100            raise

A JSON deserializer for classes which implements the [Deserializable] interface.

Further reading: https://github.com/LostInDarkMath/pedantic-python-decorators/issues/55

GenericFlaskDeserializer( cls: type[Deserializable], catch_exception: bool = True, **kwargs: Any)
81    def __init__(self, cls: type[Deserializable], catch_exception: bool = True, **kwargs: Any) -> None:  # noqa: D107
82        super().__init__(**kwargs)
83        self._cls = cls
84        self._catch_exceptions = catch_exception
@overrides(ExternalParameter)
def has_value(self) -> bool:
86    @overrides(ExternalParameter)
87    def has_value(self) -> bool:  # noqa: D102
88        return request.is_json

Returns True if the value can be fetched.

@overrides(ExternalParameter)
def load_value(self) -> Any:
 90    @overrides(ExternalParameter)
 91    def load_value(self) -> Any:  # noqa: D102
 92        try:
 93            return self._cls.from_json(request.json)
 94        except ValidatorException as ex:
 95            raise ParameterException.from_validator_exception(exception=ex, parameter_name='') from ex
 96        except Exception as ex:
 97            if self._catch_exceptions:
 98                self.raise_exception(msg=str(ex))
 99
100            raise

Loads a value and returns it.