Examples

Function decorator

Validobj can be used in a function decorator to automatically validate each function argument based on its type annotation. The code to achieve is:

#autoparse_decorator.py

import functools
from typing import get_type_hints
import inspect

from validobj import parse_input, ValidationError


def autoparse(f):
    """Apply ``parse_input`` to each argument of the decorated function, based
    on its type annotation."""
    sig = inspect.signature(f)
    # Use this instead of signature to resolve delayed annotations.
    hints = get_type_hints(f)

    @functools.wraps(f)
    def f_(*args, **kwargs):
        ba = sig.bind(*args, **kwargs)
        newargs = ba.arguments.copy()
        for argname, argvalue in ba.arguments.items():
            param = sig.parameters[argname]
            if argname not in hints:
                continue
            spec = hints[argname]
            try:
                # Handle variable keyword arguments. The type annotation applies to each value.
                if param.kind is param.VAR_KEYWORD:
                    newargs[argname] = {
                        k: parse_input(v, spec) for k, v in argvalue.items()
                    }

                # Handle variable positional arguments. The type annotation applies to each value.
                elif param.kind is param.VAR_POSITIONAL:
                    newargs[argname] = tuple(parse_input(v, spec) for v in argvalue)
                else:
                    newargs[argname] = parse_input(argvalue, spec)
            except ValidationError as e:
                # Add information on which argument failed
                raise ValidationError(f"Failed processing argument {argname!r}") from e

        ba.arguments = newargs

        return f(*ba.args, **ba.kwargs)

    return f_

The functions decorated with autoparse take simple input and use validobj.validation.parse_input() to make it conform to the type specification.

>>> import enum
>>> class Currency(enum.Enum):
...     EUR = 0.01
...     GBP = 0.018
...
>>> @autoparse
... def print_funds_in_euro(quantity_cents: int, currency: Currency):
...     print(f"{quantity_cents*currency.value:.2f} {Currency.EUR.name}")
>>> print_funds_in_euro(4, 'GBP')
0.07 EUR

This can be useful to perform application specific validation that cannot be easily encoded in types. For example we may want to check if a given account exists and has enough funds to perform a transfer:

import dataclasses

from validobj.errors import WrongFieldError

from autoparse_decorator import autoparse

user_funds = {'Alice': 300, 'Bob': 150, 'Eve': 33}


@dataclasses.dataclass
class Transfer:
    origin: str
    destination: str
    quantity: int


@autoparse
def check_transfer(tr: Transfer):
    if tr.origin not in user_funds:
        raise WrongFieldError("Originating user does not exist", wrong_field='origin')
    if tr.destination not in user_funds:
        raise WrongFieldError(
            "Destination user does not exist", wrong_field='destination'
        )
    if tr.quantity > user_funds[tr.origin]:
        raise WrongFieldError("Insufficient funds", wrong_field='quantity')
    return tr

Then the usage is:

>>> check_transfer({'origin': 'Bob', 'destination': 'Alice', 'quantity': 100})
Transfer(origin='Bob', destination='Alice', quantity=100)
>>> check_transfer({'origin': 'Bob', 'destination': 'Alice', 'quantity': 400})
Traceback (most recent call last):
    ...
WrongFieldError: Insufficient funds

YAML line numbers

The errors in Validobj provide enough information to associate the line number with the cause of the error, when combined with a library that tracks the line information such as ruamel.yaml. It is possible to climb up the __cause__ of the errors to produce a detailed traceback. The following code achieves that:

# yaml_processing.py


from validobj import parse_input, ValidationError


def parse_yaml_inp(inp, spec):
    try:
        return parse_input(inp, spec)
    except ValidationError as e:
        current_exc = e
        current_inp = inp
        error_text_lines = []
        while current_exc:
            if hasattr(current_exc, 'wrong_field'):
                wrong_field = current_exc.wrong_field
                # Mappings compping from ``round_trip_load`` have an
                # ``lc`` attribute that gives a tuple of
                # ``(line_number, column)`` for a given item in
                # the mapping.
                line = current_inp.lc.item(wrong_field)[0]
                error_text_lines.append(f"Problem processing key at line {line}:")
                current_inp = current_inp[wrong_field]
            elif hasattr(current_exc, 'wrong_index'):
                wrong_index = current_exc.wrong_index
                # Similarly lists allow to retrieve the line number for
                # a given item.
                line = current_inp.lc.item(wrong_index)[0]
                current_inp = current_inp[wrong_index]
                error_text_lines.append(f"Problem processing list item at line {line}:")
            elif hasattr(current_exc, 'unknown'):
                unknown_lines = []
                for u in current_exc.unknown:
                    unknown_lines.append((current_inp.lc.item(u)[0], u))
                unknown_lines.sort()
                for line, key in unknown_lines:
                    error_text_lines.append(
                        f"Unknown key {key!r} defined at line {line}:"
                    )
            error_text_lines.append(str(current_exc))
            current_exc = current_exc.__cause__
        raise ValidationError('\n'.join(error_text_lines)) from e

An example usage is:

import sys
import dataclasses
import enum
from typing import Mapping, Set, Tuple, List

import ruamel.yaml as yaml

from yaml_processing import parse_yaml_inp, ValidationError


class DiskPermissions(enum.Flag):
    READ = enum.auto()
    WRITE = enum.auto()
    EXECUTE = enum.auto()


class OS(enum.Enum):
    mac = enum.auto()
    windows = enum.auto()
    linux = enum.auto()


@dataclasses.dataclass
class Job:
    name: str
    os: Set[OS]
    script_path: str
    framework_version: Tuple[int, int] = (1, 0)
    disk_permissions: DiskPermissions = DiskPermissions.READ


@dataclasses.dataclass
class CIConf:
    stages: List[Job]
    global_environment: Mapping[str, str] = dataclasses.field(default_factory=dict)


inp = yaml.round_trip_load(
    """
global_environment: {"CI_ACTIVE": "1"}
stages:
  - name: compile
    os: [linux, mac]
    script_path: "build.sh"
  - name: test
    os: [linux, mac]
    script: "test.sh"
"""
)




try:
    parse_yaml_inp(inp, CIConf)
except ValidationError as e:
    print(e)
    sys.exit(1)

Which prints:

JSON NaNs and Infinities

The JSON specification does not allow for NaN or infinity floating point values, which is occasionally inconvenient. We might want to introduce the convention that the JSON strings "Infinity", "-Infinity" and "NaN", in addition to all integer and floating point values, should resolve to float values. A type annotated Python function that does that is

>>> from typing import Union, Literal
>>> def json_float(
...     inp: Union[int, float, Literal["Infinity"], Literal["-Infinity"], Literal["NaN"]]
... ) -> float:
...     return float(inp)

(since the three literal strings work with the built in float).

The custom parsing feature allows passing that function to

>>> from validobj.custom import Parser
>>> JSONFloat = Parser(json_float)

This reads the function signature to construct a valid typing.Annotated annotation. It then allows associating the input type of the function (the union) with its output type (float), and invokes it to go from one to the other at arbitrary nested levels. For example we might want to define an object allowing for infinities as

>>> from typing import List
>>> import dataclasses
>>> from validobj import parse_input, ValidationError
>>> @dataclasses.dataclass
... class Bin:
...     min_value: JSONFloat
...     max_value: JSONFloat
...     def __post_init__(self):
...         if not self.min_value <= self.max_value:
...             raise ValidationError("Expecting min_value <= max_value")
>>> Binning = List[Bin]

>>> parse_input(
...     [{"min_value": "-Infinity", "max_value": 5}, {"min_value": 0, "max_value": 10}],
...     Binning
... )
[Bin(min_value=-inf, max_value=5.0), Bin(min_value=0.0, max_value=10.0)]