dlt.common.destination.utils
MERGE_CONDITION_PLACEHOLDERS
Placeholders that expand to table identifiers.
MERGE_CONDITION_VALUE_PLACEHOLDERS
Placeholders that dlt fills from the load package. They expand to literals.
validate_merge_condition
def validate_merge_condition(
hint_name: str,
filter_: str,
placeholders: Sequence[str] = MERGE_CONDITION_PLACEHOLDERS) -> None
Raises ValueError if the merge condition filter_ is malformed or uses a placeholder
not in placeholders.
get_merge_placeholder_values
def get_merge_placeholder_values(table_name: str) -> Dict[str, Any]
Returns the values of the value placeholders of table_name in the current load package.
Returns an empty dict outside of a load package.
to_literal_datetime
def to_literal_datetime(value: TAnyDateTime,
supports_tz_aware_datetime: bool = True,
precision: int = 6) -> datetime
Returns value as the destination stores a loaded timestamp: in the context timezone, as
the naive wall clock on a destination without tz support, at precision. A naive value is
read in the context timezone.
get_scd2_timestamps
def get_scd2_timestamps(
root_table: TTableSchema, caps: DestinationCapabilitiesContext
) -> Tuple[datetime, Optional[datetime]]
Returns the boundary and active record timestamps of an scd2 table as literal values of
the destination, see to_literal_datetime. The boundary comes from the load package.
expand_merge_condition
def expand_merge_condition(hint_name: str,
filter_: str,
identifiers: Dict[str, str],
values: Dict[str, Any],
escape_literal: Callable[[Any], Any],
supports_tz_aware_datetime: bool = True,
timestamp_precision: int = 6) -> str
Expands identifier placeholders as they are and value placeholders as escaped literals.
A datetime value is converted with to_literal_datetime. A list or tuple value expands
to a list of literals for IN, an empty one to (NULL).
Raises ValueError if filter_ uses a placeholder that has no identifier or value.
get_merge_compare_columns
def get_merge_compare_columns(table: TTableSchema) -> List[str]
Returns names of columns whose values decide whether a merged row changed.
Returns only the row version column when the table has one. Otherwise excludes the key
columns, the hard_delete column and _dlt_load_id.
get_merge_changed_cond
def get_merge_changed_cond(
table: TTableSchema,
source_alias: str,
target_alias: str,
escape_id: Callable[[str], str] = str) -> Optional[str]
Returns a SQL condition that holds when a source row differs from its target row.
Returns None when the table has no columns to compare.
verify_schema_capabilities
def verify_schema_capabilities(schema: Schema,
capabilities: DestinationCapabilitiesContext,
destination_type: str,
warnings: bool = True) -> List[Exception]
Verifies load_tables that have all hints filled by job client before loading against capabilities.
Returns a list of exceptions representing critical problems with the schema.
It will log warnings by default. It is up to the caller to eventually raise exception
- Checks all table and column name lengths against destination capabilities and raises on too long identifiers
- Checks if schema has collisions due to case sensitivity of the identifiers
column_type_to_str
def column_type_to_str(column: TColumnType) -> str
Converts column type to db-like type string
prepare_load_table
@with_config
def prepare_load_table(
tables: TSchemaTables,
table: TTableSchema,
destination_capabilities: DestinationCapabilitiesContext = ConfigValue
) -> PreparedTableSchema
Prepares a table schema to be loaded by filling missing hints and doing other modifications required by given destination.
destination_capabilities are injected from context if not explicitly passed.
Returns: prepared table, note: table is cloned
resolve_replace_strategy
def resolve_replace_strategy(
table: PreparedTableSchema,
required_strategy: Optional[TLoaderReplaceStrategy],
destination_capabilities: Optional[DestinationCapabilitiesContext]
) -> Optional[TLoaderReplaceStrategy]
Returns replace strategy for the supplied prepared table and destination capabilities.
If required_strategy cannot be fulfilled or capabilities do not support replace, None is returned.
resolve_merge_strategy
@with_config
def resolve_merge_strategy(
tables: TSchemaTables,
table: TTableSchema,
destination_capabilities: Optional[
DestinationCapabilitiesContext] = ConfigValue,
*,
is_prepared: bool) -> Optional[TLoaderMergeStrategy]
Resolve merge strategy for a table, possibly resolving the 'x-merge-strategy from a table chain. strategies selector in destination_capabilities
is used if present. If table does not contain strategy hint, a default value will be used which is the first.
destination_capabilities are injected from context if not explicitly passed. A table
prepared by a job client (is_prepared) is used as it is, other tables are prepared from
tables.
Returns None if table write disposition is not merge