common.typing

StrAny

immutable, covariant entity

StrStr

immutable, covariant entity

StrStrStr

immutable, covariant entity

TFun

any function

TSecretValue

type: ignore

TSecretStrValue

type: ignore

TDataItem

A single data item as extracted from data source

TDataItems

A single data item or a list as extracted from the data source

TAnyDateTime

DateTime represented as pendulum/python object, ISO string or unix timestamp

TLoaderFileFormat

known loader file formats

ConfigValueSentinel Objects

class ConfigValueSentinel(NamedTuple)

[view_source]

Class to create singleton sentinel for config and secret injected value

ConfigValue

Config value indicating argument that may be injected by config provider. Evaluates to None when type checking

SecretValue

Secret value indicating argument that may be injected by config provider. Evaluates to None when type checking

SupportsVariant Objects

@runtime_checkable
class SupportsVariant(Protocol, Generic[TVariantBase])

[view_source]

Defines variant type protocol that should be recognized by normalizers

Variant types behave like TVariantBase type (ie. Decimal) but also implement the protocol below that is used to extract the variant value from it. See Wei type declaration which returns Decimal or str for values greater than supported by destination warehouse.

SupportsHumanize Objects

class SupportsHumanize(Protocol)

[view_source]

asdict

def asdict() -> DictStrAny

[view_source]

Represents object as dict with a schema loadable by dlt

asstr

def asstr(verbosity: int = 0) -> str

[view_source]

Represents object as human readable string

get_type_name

def get_type_name(t: Type[Any]) -> str

[view_source]

Returns a human-friendly name of type t

is_callable_type

def is_callable_type(hint: Type[Any]) -> bool

[view_source]

Checks if hint is callable: a function or callable class. This function does not descend into type arguments ie. if Union, Literal or NewType contain callables, those are ignored

get_literal_args

def get_literal_args(literal: Type[Any]) -> List[Any]

[view_source]

Recursively get arguments from nested Literal types and return an unified list.

extract_inner_type

def extract_inner_type(hint: Type[Any],
                       preserve_new_types: bool = False,
                       preserve_literal: bool = False) -> Type[Any]

[view_source]

Gets the inner type from Literal, Optional, Final and NewType

Arguments:

• hint Type[Any] - Type to extract
• preserve_new_types bool - Do not extract supertype of a NewType

Returns:

• Type[Any] - Inner type if hint was Literal, Optional or NewType, otherwise hint

get_all_types_of_class_in_union

def get_all_types_of_class_in_union(hint: Any,
                                    cls: TAny,
                                    with_superclass: bool = False
                                    ) -> List[TAny]

[view_source]

if hint is an Union that contains classes, return all classes that are a subclass or (optionally) superclass of cls

is_generic_alias

def is_generic_alias(tp: Any) -> bool

[view_source]

Tests if type is a generic alias ie. List[str]

is_subclass

def is_subclass(subclass: Any, cls: Any) -> bool

[view_source]

Return whether 'cls' is a derived from another class or is the same class.

Will handle generic types by comparing their origins.

get_generic_type_argument_from_instance

def get_generic_type_argument_from_instance(instance: Any,
                                            sample_value: Optional[Any] = None
                                            ) -> Type[Any]

[view_source]

Infers type argument of a Generic class from an instance of that class using optional sample_value of the argument type

Inference depends on the presence of orig_class attribute in instance, if not present - sample_Value will be used

Arguments:

• instance Any - instance of Generic class
• sample_value Optional[Any] - instance of type of generic class, optional

Returns:

• Type[Any] - type argument or Any if not known

copy_sig

def copy_sig(
    wrapper: Callable[TInputArgs, Any]
) -> Callable[[Callable[..., TReturnVal]], Callable[TInputArgs, TReturnVal]]

[view_source]

Copies docstring and signature from wrapper to func but keeps the func return value type

copy_sig_any

def copy_sig_any(
    wrapper: Callable[Concatenate[TDataItem, TInputArgs], Any]
) -> Callable[[Callable[..., TReturnVal]], Callable[Concatenate[
        TDataItem, TInputArgs], TReturnVal]]

[view_source]

Copies docstring and signature from wrapper to func but keeps the func return value type

It converts the type of first argument of the wrapper to Any which allows to type transformers in DltSources. See filesystem source readers as example