Warning: This document is for an old version of Pydra: A simple dataflow engine with scalable semantics. The main version is master.

pydra.utils.typing module

class pydra.utils.typing.TypeParser(tp, coercible: ~typing.Iterable[~typing.Tuple[type | ~typing.Any, type | ~typing.Any]] | None = ((typing.Sequence, typing.Sequence), (typing.Mapping, typing.Mapping), (<class 'pathlib.Path'>, <class 'os.PathLike'>), (<class 'str'>, <class 'os.PathLike'>), (<class 'os.PathLike'>, <class 'pathlib.Path'>), (<class 'os.PathLike'>, <class 'str'>), (typing.Any, <class 'pydra.engine.specs.MultiInputObj'>), (<class 'int'>, <class 'float'>)), not_coercible: ~typing.Iterable[~typing.Tuple[type | ~typing.Any, type | ~typing.Any]] | None = ((<class 'str'>, typing.Sequence), (typing.Sequence, <class 'str'>)))

Bases: Generic[T]

A callable which can be used as a converter for attrs.fields to check whether an object or LazyField matches the specified field type, or can be coerced into it (given the criteria passed on initialisation of the checker). Nested container type are expanded and each of their type args are checked/coerced against corresponding parts of the object.

Parameters:
  • tp (type) – the type objects will be coerced to

  • coercible (Iterable[ty.Tuple[type or Any, type or Any]], optional) – limits coercing between the pairs of types where they appear within the tree of more complex nested container types. If None, then all types are coercible except explicitly excluded

  • not_coercible (Iterable[ty.Tuple[type or Any, type or Any]], optional) – excludes the limits coercing between the pairs of types where they appear within the tree of more complex nested container types. Overrides ‘coercible’ to enable you to carve out exceptions, such as TypeParser(list, coercible=[(ty.Iterable, list)], not_coercible=[(str, list)])

COERCIBLE_DEFAULT: Tuple[Tuple[type, type], ...] = ((typing.Sequence, typing.Sequence), (typing.Mapping, typing.Mapping), (<class 'pathlib.Path'>, <class 'os.PathLike'>), (<class 'str'>, <class 'os.PathLike'>), (<class 'os.PathLike'>, <class 'pathlib.Path'>), (<class 'os.PathLike'>, <class 'str'>), (typing.Any, <class 'pydra.engine.specs.MultiInputObj'>), (<class 'int'>, <class 'float'>))
NOT_COERCIBLE_DEFAULT = ((<class 'str'>, typing.Sequence), (typing.Sequence, <class 'str'>))
classmethod apply_to_instances(target_type: Type[Any], func: Callable, value: Any, cache: Dict[int, Any] | None = None) Any

Applies a function to all instances of the given type that are potentially nested within the given value, caching previously computed modifications to handle repeated elements

Parameters:
  • target_type (type) – the target type to apply the function to

  • func (callable) – the callable object (e.g. function) to apply to the instances

  • value (Any) – the value to copy files from (if required)

  • cache (dict, optional) – guards against multiple references to the same objects by keeping a cache of the modified

check_coercible(source: object | type, target: type | Any)

Checks whether the source object or type is coercible to the target type given the coercion rules defined in the coercible and not_coercible attrs

Parameters:
  • source (object or type) – source object or type to be coerced

  • target (type or ty.Any) – target type for the source to be coerced to

Raises:

TypeError – If the source type cannot be coerced into the target type depending on the explicit inclusions and exclusions set in the coercible and not_coercible member attrs

check_type(type_: Type[Any])

Checks the given type to see whether it matches or is a subtype of the specified type or whether coercion rule is specified between the types

Parameters:

type_ (ty.Type[ty.Any]) – the type to check whether it is coercible into the specified type

Raises:

TypeError – if the type is not either the specified type, a sub-type or coercible to it

coerce(object_: Any) T

Attempts to coerce the given object to the type of the specified type

coercible: List[Tuple[type | Any, type | Any]]
classmethod contains_type(target: Type[Any], type_: Type[Any])

Checks a potentially nested type for sub-classes of the target type

Parameters:
  • target (type) – the target type to check for sub-classes of

  • type_ (type) – the type to check for nested types that are sub-classes of target

static get_args(tp)

Get type arguments with all substitutions performed.

For unions, basic simplifications used by Union constructor are performed.

Examples:

>>> T = TypeVar('T')
>>> assert get_args(Dict[str, int]) == (str, int)
>>> assert get_args(int) == ()
>>> assert get_args(Union[int, Union[T, int], str][int]) == (int, str)
>>> assert get_args(Union[int, Tuple[T, int]][str]) == (int, Tuple[str, int])
>>> assert get_args(Callable[[], T][int]) == ([], int)
classmethod get_item_type(sequence_type: Type[Sequence[T]]) Type[T] | Any

Return the type of the types of items in a sequence type

Parameters:

sequence_type (type[Sequence]) – the type to find the type of the items of

Returns:

item_type – the type of the items

Return type:

type or None

static get_origin(tp)

Get the unsubscripted version of a type.

This supports generic types, Callable, Tuple, Union, Literal, Final, ClassVar, Annotated, and others. Return None for unsupported types.

Examples:

>>> P = ParamSpec('P')
>>> assert get_origin(Literal[42]) is Literal
>>> assert get_origin(int) is None
>>> assert get_origin(ClassVar[int]) is ClassVar
>>> assert get_origin(Generic) is Generic
>>> assert get_origin(Generic[T]) is Generic
>>> assert get_origin(Union[T, int]) is Union
>>> assert get_origin(List[Tuple[T, T]][int]) is list
>>> assert get_origin(P.args) is P
static is_instance(obj: object, candidates: Type[Any] | Iterable[Type[Any]]) bool

Checks whether the object is an instance of cls or that cls is typing.Any, extending the built-in isinstance to check nested type args

Parameters:
  • obj (object) – the object to check whether it is an instance of one of the candidates

  • candidates (type or ty.Iterable[type]) – the candidate types to check the object against

classmethod is_subclass(klass: Type[Any], candidates: Type[Any] | Iterable[Type[Any]], any_ok: bool = False) bool

Checks whether the class a is either the same as b, a subclass of b or b is typing.Any, extending built-in issubclass to check nested type args

Parameters:
  • klass (type) – the klass to check whether it is a subclass of one of the candidates

  • candidates (type or ty.Iterable[type]) – the candidate types to check the object against

  • any_ok (bool) – whether klass=typing.Any should return True or False

classmethod matches(obj: Type[Any], target: Type[Any], **kwargs) bool

Returns true if the provided type matches the pattern of the TypeParser

Parameters:
  • type_ (type) – the type to check

  • target (type) – the target type to check against

  • **kwargs (dict[str, Any], optional) – passed on to TypeParser.__init__

Returns:

matches – whether the type matches the target type factoring in sub-classes and coercible pairs

Return type:

bool

classmethod matches_type(type_: Type[Any], target: Type[Any], **kwargs) bool

Returns true if the provided type matches the pattern of the TypeParser

Parameters:
  • type_ (type) – the type to check

  • target (type) – the target type to check against

  • **kwargs (dict[str, Any], optional) – passed on to TypeParser.__init__

Returns:

matches – whether the type matches the target type factoring in sub-classes and coercible pairs

Return type:

bool

not_coercible: List[Tuple[type | Any, type | Any]]
classmethod strip_splits(type_: Type[Any]) Tuple[Type, int]

Strips any StateArray types from the outside of the specified type and returns the stripped type and the depth it was found at

Parameters:
  • type_ (ty.Type[ty.Any]) – the type to list the nested sequences of

  • only_splits (bool, optional) – whether to only return nested splits, not all sequence types

Returns:

  • inner_type (type) – the inner type once all outer sequences are stripped

  • depth (int) – the number of splits outside the inner_type

tp: Type[T]