dlt.common.typing
CallableAny
A special callable Any that returns argument but is recognized as Any type by dlt hint checkers
StrAny
immutable, covariant entity
StrStr
immutable, covariant entity
StrStrStr
immutable, covariant entity
TFun
any function
SecretSentinel Objects
class SecretSentinel()
Marks a secret type when part of type annotations
TColumnNames
A string representing a column name or a sequence of
TTableNames
A string representing a table name or a sequence of
TDataItem
A single data item as extracted from data source
TDataItems
A single data item or a list as extracted from the data source
TDataRecord
Table row dictionary. Not guaranteed to be JSON serializable without custom encoding.
TDataRecordBatch
List of table row dictionaries. Not guaranteed to be JSON serializable without custom encoding.
TAnyDateTime
DateTime represented as pendulum/python object, ISO string or unix timestamp
TTimeInterval
Half-open time interval [start, end) as timezone-aware datetimes.
TLoaderFileFormat
known loader file formats
ConfigValueSentinel Objects
class ConfigValueSentinel(NamedTuple)
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])
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)
asdict
def asdict() -> DictStrAny
Represents object as dict with a schema loadable by dlt
asstr
def asstr(verbosity: int = 0) -> str
Represents object as human readable string
SocketLike Objects
class SocketLike(Protocol)
send
def send(data: bytes) -> int
Send bytes; return number of bytes sent.
recv
def recv(bufsize: int) -> bytes
Receive up to bufsize bytes and return them.
close
def close() -> None
Close the socket-like connection.
settimeout
def settimeout(timeout: float) -> None
Set the timeout (in seconds) for send/recv operations.
get_type_name
def get_type_name(t: Type[Any]) -> str
Returns a human-friendly name of type t
is_callable_type
def is_callable_type(hint: Type[Any]) -> bool
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
is_any_type
def is_any_type(t: Type[Any]) -> bool
Checks if t is one of recognized Any types
get_literal_args
def get_literal_args(literal: Type[Any]) -> List[Any]
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,
preserve_annotated: bool = False) -> Type[Any]
Gets the inner type from Literal, Optional, Final and NewType
Arguments:
hintType[Any] - Type to extractpreserve_new_typesbool - 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]
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
Tests if type is a generic alias ie. List[str]
is_subclass
def is_subclass(subclass: Any, cls: Any) -> bool
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]
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:
instanceAny - instance of Generic classsample_valueOptional[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]]
Copies docstring and signature from wrapper to func but keeps the func return value type
copy_sig_ret
def copy_sig_ret(
wrapper: Callable[TInputArgs, Any], ret: Type[TReturnVal]
) -> Callable[[Callable[TInputArgs, Any]], Callable[TInputArgs, TReturnVal]]
Copies docstring and signature from wrapper to func and replaces return type with ret
copy_sig_any
def copy_sig_any(
wrapper: Callable[Concatenate[TDataItem, TInputArgs], Any]
) -> Callable[[Callable[..., TReturnVal]], Callable[Concatenate[
TDataItem, TInputArgs], TReturnVal]]
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
add_value_to_literal
def add_value_to_literal(literal: Any, value: Any) -> None
Extends a Literal at runtime with a new value.
Arguments:
literalType[Any] - Literal to extendvalueAny - Value to add
get_type_globals
def get_type_globals(obj: Any) -> Dict[str, Any]
Best-effort extraction of globals() associated with a type. If object is passed, we get its class
Handles:
- functions
- classes (including TypedDict, dataclasses, Pydantic models, etc.)
- modules (returns their dict)
resolve_single_annotation
def resolve_single_annotation(ann: Any,
*,
globalns: Optional[Dict[str, Any]] = None,
localns: Optional[Dict[str, Any]] = None,
raise_on_error: bool = False) -> Any
Resolves annotation ann if it is a str and/or ForwardRef.
- If
annis not a str or ForwardRef, it's returned unchanged. - If it is a str/ForwardRef, we eval it in an appropriate namespace.