dlt.extract.utils
NORMALIZE_PARALLEL_MIN_BYTES
Data to normalize below this size takes seconds, so idle normalize workers are not reported
NORMALIZE_PARALLEL_MIN_USAGE
Files are reported as unbalanced when normalize would use at most this share of its workers
get_data_item_format
def get_data_item_format(items: TDataItems) -> TDataItemFormat
Detect the format of the data item from items.
Reverts to object for empty lists
Returns:
The data file format.
get_arrow_fields
def get_arrow_fields(items: TDataItems) -> Dict[str, "pa.Field"]
Returns arrow fields of items by name, later items override earlier ones.
resolve_column_value
def resolve_column_value(column_hint: TTableHintTemplate[TColumnNames],
item: TDataItem) -> Union[Any, List[Any]]
Extract values from the data item given a column hint. Returns either a single value or list of values when hint is a composite.
DEDUP_HASH_BYTES
Digest length of an incremental dedup hash, 12 characters in base64.
LEGACY_DEDUP_HASH_LEN
Length of a dedup hash written before 1.29: a 15 byte digest over the Z rendering of UTC.
digest_dedup_value
def digest_dedup_value(value: Any, legacy: bool = False) -> str
Stable content hash of a JSON-serializable value used for incremental dedup.
legacy reproduces the hash written before 1.29, so hashes kept in older state still match.
has_legacy_dedup_hashes
def has_legacy_dedup_hashes(hashes: Iterable[str]) -> bool
True when hashes were written before 1.29 and only match legacy digests.
ensure_table_schema_columns
def ensure_table_schema_columns(
columns: TAnySchemaColumns) -> TTableSchemaColumns
Convert supported column schema types to a column dict which can be used in resource schema.
Arguments:
columns- A dict of column schemas, a list of column schemas, a pydantic model, or a pyarrow schema
ensure_table_schema_columns_hint
def ensure_table_schema_columns_hint(
columns: TTableHintTemplate[TAnySchemaColumns]
) -> TTableHintTemplate[TTableSchemaColumns]
Convert column schema hint to a hint returning TTableSchemaColumns.
A callable hint is wrapped in another function which converts the original result.
simulate_func_call
def simulate_func_call(
f: Union[Any, AnyFun], args_to_skip: int, *args: Any, **kwargs: Any
) -> Tuple[inspect.Signature, inspect.Signature, inspect.BoundArguments]
Simulates a call to a resource or transformer function before it will be wrapped for later execution in the pipe
Returns a tuple with a f signature, modified signature in case of transformers and bound arguments
wrap_iterator
def wrap_iterator(gen: Iterator[TDataItems]) -> Iterator[TDataItems]
Wraps an iterator into a generator
wrap_async_iterator
def wrap_async_iterator(
gen: AsyncIterator[TDataItems]
) -> Generator[Awaitable[TDataItems], None, None]
Wraps an async generator into a list of awaitables
wrap_parallel_iterator
def wrap_parallel_iterator(f: TAnyFunOrGenerator) -> TAnyFunOrGenerator
Wraps a generator for parallel extraction
wrap_compat_transformer
def wrap_compat_transformer(name: str, f: AnyFun, sig: inspect.Signature,
*args: Any, **kwargs: Any) -> AnyFun
Creates a compatible wrapper over transformer function. A pure transformer function expects data item in first argument and one keyword argument called meta
wrap_resource_gen
def wrap_resource_gen(name: str, f: AnyFun, sig: inspect.Signature, *args: Any,
**kwargs: Any) -> AnyFun
Wraps a generator or generator function so it is evaluated on extraction and binds argument passed to call
make_schema_with_default_name
def make_schema_with_default_name(pipeline_name: str) -> str
Makes schema name from the pipeline name using the name normalizer. "_pipeline" suffix is removed if present
get_unbalanced_files_warning
def get_unbalanced_files_warning(pipeline_name: str, load_id: str,
files: Sequence[DataWriterMetrics],
file_limits: Mapping[str,
Tuple[Optional[int],
Optional[int]]],
dlt_tables: Collection[str],
workers: int) -> Optional[str]
Returns a warning if files leave at least half of the normalize workers capacity idle.
file_limits maps table names to file_max_items and file_max_bytes of their writers.
dynstr Objects
class dynstr(str)
Dynamic string which will generate final value when called
get_code_ref
def get_code_ref(data: Any) -> Optional[str]
Returns the fully qualified name of the function data is or was created from.
Returns None for data that is not a function or a generator, and for anonymous code like
lambdas or generator expressions, whose names are not unique within a module.