dlt.common.destination.client
StorageSchemaInfo Objects
class StorageSchemaInfo(NamedTuple)
from_normalized_mapping
@classmethod
def from_normalized_mapping(
cls, normalized_doc: Dict[str, Any],
naming_convention: NamingConvention) -> "StorageSchemaInfo"
Instantiate this class from mapping where keys are normalized according to given naming convention
Arguments:
normalized_doc- Mapping with normalized keys (e.g. {Version: ..., SchemaName: ...})naming_convention- Naming convention that was used to normalize keys
Returns:
StorageSchemaInfo- Instance of this class
to_normalized_mapping
def to_normalized_mapping(
naming_convention: NamingConvention) -> Dict[str, Any]
Convert this instance to mapping where keys are normalized according to given naming convention
Arguments:
naming_convention- Naming convention that should be used to normalize keys
Returns:
Dict[str, Any]: Mapping with normalized keys (e.g. {Version: ..., SchemaName: ...})
StateInfo Objects
@dataclasses.dataclass
class StateInfo()
from_normalized_mapping
@classmethod
def from_normalized_mapping(
cls, normalized_doc: Dict[str, Any],
naming_convention: NamingConvention) -> "StateInfo"
Instantiate this class from mapping where keys are normalized according to given naming convention
Arguments:
normalized_doc- Mapping with normalized keys (e.g. {Version: ..., PipelineName: ...})naming_convention- Naming convention that was used to normalize keys
Returns:
StateInfo- Instance of this class
DestinationClientConfiguration Objects
@configspec
class DestinationClientConfiguration(BaseConfiguration)
destination_type
which destination to load data to
destination_name
name of the destination
data_location
def data_location() -> Optional[str]
Returns the data location that the query engine of this destination can access with the supplied credentials.
The location is limited to data that the engine accesses without an additional attach to another location and without federation.
Returns:
Optional[str]-Nonewhen the destination has no data location, for example a reverse ETL sink. Such a destination has no query engine, so it accesses no data.
Raises:
ConfigurationValueError- When the location cannot be computed from this configuration.
fingerprint
def fingerprint() -> str
Returns a destination fingerprint derived from selected configuration fields.
is_same_location
def is_same_location(other: "DestinationClientConfiguration") -> bool
Checks if self and other have the same destination type. Also checks if they
store data in the same location.
Raises:
ConfigurationValueError- When either location cannot be computed.
can_read_from
def can_read_from(other: "DestinationClientConfiguration") -> bool
Returns True if self can read data from other.
In case of SQL engines it is an ability to SELECT / JOIN
This check must be as wide as possible. On many destinations it describes all data that the query engine joins together. A destination that federates or attaches data locations also includes that data, when it implements the attach interface. Only duckdb implements the interface now.
can_write_from
def can_write_from(other: "DestinationClientConfiguration") -> bool
Returns true if self can write data from other
In case of SQL engines it is an ability to INSERT FROM
__str__
def __str__() -> str
Return displayable destination location
credentials_type
@classmethod
def credentials_type(
cls,
config: "DestinationClientConfiguration" = None
) -> Type[CredentialsConfiguration]
Figure out credentials type, using hint resolvers for dynamic types
For correct type resolution of filesystem, config should have bucket_url populated
WithAttachableEngine Objects
class WithAttachableEngine()
Marks a destination with a query engine that attaches foreign datasets. A foreign engine can attach the datasets of this destination in turn.
This class must come first in the list of mixins, so that can_read_from here overrides
the base class.
attach_type
def attach_type() -> Optional["TAttachType"]
Returns the engine type that can run the attach statements of this destination.
can_attach
def can_attach(attach_type: "TAttachType") -> bool
Checks if the engine of this destination can run attach_type statements.
needs_attach
def needs_attach(other: DestinationClientConfiguration) -> bool
Checks if this query engine must attach other, or if it already accesses the data
of other.
can_read_from
def can_read_from(other: DestinationClientConfiguration) -> bool
Returns True when this query engine already accesses other, or can attach it.
DestinationClientDwhConfiguration Objects
@configspec
class DestinationClientDwhConfiguration(DestinationClientConfiguration)
Configuration of a destination that supports datasets/schemas
dataset_name
dataset name in the destination to load data to, for schemas that are not default schema, it is used as dataset prefix
default_schema_name
name of default schema to be used to name effective dataset to load data to
replace_strategy
How to handle replace disposition for this destination, uses first strategy from caps if not declared
staging_dataset_name_layout
Layout for staging dataset, where %s is replaced with dataset name. placeholder is optional
enable_dataset_name_normalization
Whether to normalize the dataset name. Affects staging dataset as well.
info_tables_query_threshold
Threshold for information schema tables query, if exceeded tables will be filtered in code.
normalize_dataset_name
def normalize_dataset_name(schema: Schema) -> str
Builds full db dataset (schema) name out of configured dataset name and schema name: {dataset_name}_{schema.name}. The resulting name is normalized.
If default schema name is None or equals schema.name, the schema suffix is skipped.
normalize_staging_dataset_name
def normalize_staging_dataset_name(schema: Schema) -> str
Builds staging dataset name out of dataset_name and staging_dataset_name_layout.
needs_dataset_name
@classmethod
def needs_dataset_name(cls) -> bool
Checks if configuration requires dataset name to be present. Empty datasets are allowed ie. for schema-less destinations like weaviate or clickhouse
DestinationClientStagingConfiguration Objects
@configspec
class DestinationClientStagingConfiguration(DestinationClientDwhConfiguration)
Configuration of a staging destination, able to store files with desired layout at bucket_url.
Also supports datasets and can act as standalone destination.
warn_unsafe_layout_separators
Warns when a layout separator around {table_name} can also occur inside a table name.
DestinationClientDwhWithStagingConfiguration Objects
@configspec
class DestinationClientDwhWithStagingConfiguration(
DestinationClientDwhConfiguration)
Configuration of a destination that can take data from staging destination
staging_config
configuration of the staging, if present, injected at runtime
truncate_tables_on_staging_destination_before_load
If dlt should truncate the tables on staging destination before loading data.
LoadJob Objects
class LoadJob(ABC)
A stateful load job, represents one job file
job_id
def job_id() -> str
The job id that is derived from the file name and does not changes during job lifecycle
file_name
def file_name() -> str
A name of the job file
state
@abstractmethod
def state() -> TLoadJobState
Returns current state. Should poll external resource if necessary.
failed_message
@abstractmethod
def failed_message() -> str
The error message in failed or retry states
exception
@abstractmethod
def exception() -> BaseException
The exception associated with failed or retry states
set_final_state
@abstractmethod
def set_final_state(state: TLoadJobState,
failed_message: Optional[str] = None) -> None
Forces the job into a terminal state without running it.
metrics
def metrics() -> Optional[LoadJobMetrics]
Returns job execution metrics
RunnableLoadJob Objects
class RunnableLoadJob(LoadJob, ABC)
Represents a runnable job that loads a single file
Each job starts in "running" state and ends in one of terminal states: "retry", "failed" or "completed".
Each job is uniquely identified by a file name. The file is guaranteed to exist in "running" state. In terminal state, the file may not be present.
In "running" state, the loader component periodically gets the state via status() method. When terminal state is reached, load job is discarded and not called again.
exception method is called to get error information in "failed" and "retry" states.
The __init__ method is responsible to put the Job in "running" state. It may raise LoadClientTerminalException and LoadClientTransientException to
immediately transition job into "failed" or "retry" state respectively.
__init__
def __init__(file_path: str) -> None
File name is also a job id (or job id is deterministically derived) so it must be globally unique
set_run_vars
def set_run_vars(
load_id: str,
schema: Schema,
load_table: PreparedTableSchema,
on_completed: Optional[Callable[["TLoadJobState", Optional[str]],
None]] = None
) -> None
Called by the loader right before the job is run.
set_final_state
def set_final_state(state: TLoadJobState,
failed_message: Optional[str] = None) -> None
Forces the job into a terminal state without running it.
Used when restoring a job from a pending transition — the destination already committed the data, so we skip execution.
run_managed
def run_managed(job_client: "JobClientBase",
done_event: BoundedSemaphore) -> None
wrapper around the user implemented run method
run
@abstractmethod
def run() -> None
run the actual job, this will be executed on a thread and should be implemented by the user exception will be handled outside of this function
state
def state() -> TLoadJobState
Returns current state. Should poll external resource if necessary.
FollowupJobRequest Objects
class FollowupJobRequest()
Base class for follow up jobs that should be created
new_file_path
@abstractmethod
def new_file_path() -> str
Path to a newly created temporary job file. If empty, no followup job should be created
HasFollowupJobs Objects
class HasFollowupJobs()
Adds a trait that allows to create single or table chain followup jobs
create_followup_jobs
def create_followup_jobs(
final_state: TLoadJobState) -> List[FollowupJobRequest]
Return list of jobs requests for jobs that should be created. final_state is state to which this job transits
JobClientBase Objects
class JobClientBase(ABC)
initialize_storage
@abstractmethod
def initialize_storage(
truncate_tables: Optional[Iterable[str]] = None) -> None
Prepares storage to be used ie. creates database schema or file system folder. Truncates requested tables.
is_storage_initialized
@abstractmethod
def is_storage_initialized() -> bool
Returns if storage is ready to be read/written.
drop_storage
@abstractmethod
def drop_storage() -> None
Brings storage back into not initialized state. Typically data in storage is destroyed.
verify_schema
def verify_schema(
only_tables: Iterable[str] = None,
new_jobs: Iterable[ParsedLoadJobFileName] = None
) -> List[PreparedTableSchema]
Verifies schema before loading, returns a list of verified loaded tables.
update_stored_schema
def update_stored_schema(
only_tables: Iterable[str] = None,
expected_update: TSchemaTables = None,
force: bool = False,
include_existing_tables: bool = False) -> Optional[TSchemaTables]
Updates storage to the current schema.
Implementations should not assume that expected_update is the exact difference between destination state and the self.schema. This is only the case if
destination has single writer and no other processes modify the schema.
Arguments:
only_tablesSequence[str], optional - Updates only listed tables. Defaults to None.expected_updateTSchemaTables, optional - Update that is expected to be applied to the destinationforcebool - force full schema migration regardless of previous updatesinclude_existing_tablesbool - Also adds missing columns to tables outside ofonly_tablesthat already exist in the destination.
Returns:
Optional[TSchemaTables]- Returns an update that was applied at the destination.
prepare_load_table
def prepare_load_table(table_name: str) -> PreparedTableSchema
Prepares a table schema to be loaded by filling missing hints and doing other modifications requires by given destination.
Returns: prepared table, note: table schema for table_name is cloned
create_load_job
@abstractmethod
def create_load_job(table: PreparedTableSchema,
file_path: str,
load_id: str,
restore: bool = False) -> LoadJob
Creates a load job for a particular table with content in file_path. Table is already prepared to be loaded.
prepare_load_job_execution
def prepare_load_job_execution(job: RunnableLoadJob) -> None
Prepare the connected job client for a load job, including load-specific query-tag context when supported.
create_table_chain_completed_followup_jobs
def create_table_chain_completed_followup_jobs(
table_chain: Sequence[PreparedTableSchema],
completed_table_chain_jobs: Optional[Sequence[LoadJobInfo]] = None
) -> List[FollowupJobRequest]
Creates a list of followup jobs that should be executed after a table chain is completed. Tables are already prepared to be loaded.
complete_load
@abstractmethod
def complete_load(load_id: str) -> None
Marks the load package with load_id as completed in the destination. Before such commit is done, the data with load_id is invalid.
WithStateSync Objects
class WithStateSync(ABC)
get_stored_schema
@abstractmethod
def get_stored_schema(schema_name: str = None) -> Optional[StorageSchemaInfo]
Retrieves newest schema with given name from destination storage.
If no name is provided, the newest schema found is retrieved. Returns None if the schema is not found or if the dataset/storage tables do not exist (DestinationUndefinedEntity is suppressed by SQL implementations).
get_stored_schema_by_hash
@abstractmethod
def get_stored_schema_by_hash(version_hash: str) -> StorageSchemaInfo
Retrieves the stored schema by hash.
Returns None if not found or if the dataset/storage tables do not exist (DestinationUndefinedEntity is suppressed by SQL implementations).
get_stored_state
@abstractmethod
def get_stored_state(pipeline_name: str) -> Optional[StateInfo]
Loads compressed state from destination storage.
Returns None if no state is found for the given pipeline name. Raises DestinationUndefinedEntity if the state or loads tables do not exist on the destination (e.g. pipeline never ran).
WithStagingDataset Objects
class WithStagingDataset(ABC)
Adds capability to use staging dataset and request it from the loader
with_staging_dataset
@abstractmethod
def with_staging_dataset() -> ContextManager["JobClientBase"]
Executes job client methods on staging dataset
create_dataset_names
@staticmethod
def create_dataset_names(
schema: Schema,
config: DestinationClientDwhConfiguration) -> Tuple[str, str]
Creates regular and staging dataset names for given schema and config. Raises a value error if the staging name is same as final dataset name. returns (dataset_name, staging_dataset_name)
SupportsStagingDestination Objects
class SupportsStagingDestination(ABC)
Adds capability to support a staging destination for the load
should_load_data_to_staging_dataset_on_staging_destination
def should_load_data_to_staging_dataset_on_staging_destination(
table_name: str) -> bool
If set to True, and staging destination is configured, the data will be loaded to staging dataset on staging destination instead of a regular dataset on staging destination. Currently it is used by Athena Iceberg which uses staging dataset on staging destination to copy data to iceberg tables stored on regular dataset on staging destination. The default is to load data to regular dataset on staging destination from where warehouses like Snowflake (that have their own storage) will copy data.
should_truncate_table_before_load_on_staging_destination
@abstractmethod
def should_truncate_table_before_load_on_staging_destination(
table_name: str) -> bool
If set to True, data in table will be truncated on staging destination (regular dataset). This is the default behavior which
can be changed with a config flag.
For Athena + Iceberg this setting is always False - Athena uses regular dataset to store Iceberg tables and we avoid touching it.
For Athena we truncate those tables only on "replace" write disposition.
should_drop_table_on_staging_destination
def should_drop_table_on_staging_destination(
dropped_table: TTableSchema) -> bool
Tells if dropped_table should be dropped on staging destination (regular dataset) in addition to dropping the table on
final destination. This stays False for all the destinations except Athena, non-iceberg where staging destination
holds actual data which needs to be deleted.
Note that dropped_table may not longer be present in schema. It is present only if it got recreated.
SupportsOpenTables Objects
class SupportsOpenTables(ABC)
Provides access to data stored in one of open table formats (iceberg or delta) and intended to be implemented by job clients.
get_open_table_catalog
@abstractmethod
def get_open_table_catalog(table_format: TTableFormat,
catalog_name: Optional[str] = None) -> Any
Gets the catalog that keeps tables' metadata. Currently only pyiceberg Catalog is supported
get_open_table_location
@abstractmethod
def get_open_table_location(table_format: TTableFormat,
table_name: str) -> str
Computes location in which table is stored which is typically a "folder" with table data and metadata. Does not verify if table exists.
Returns:
str- fully formed url with table location
load_open_table
@abstractmethod
def load_open_table(table_format: TTableFormat, table_name: str,
**kwargs: Any) -> Any
Loads table table_name metadata via catalog or directly and returns populated and authenticated
table client. Currently pyiceberg Table or DeltaTable is returned.
- table must be present in schema of job client
- table must physically exist in storage
- table may be present in associated catalog and may be automatically registered if destination configuration allows for that
- otherwise table is not found
raised DestinationUndefinedEntity if table not found
is_open_table
@abstractmethod
def is_open_table(table_format: TTableFormat, table_name: str) -> bool
Checks if table_name is stored with open table format table_format. Does not load table. Does not check if
table exists
SqlModel Objects
class SqlModel()
A SQL query with the dialect that parses it. A model also carries optional attach info for the foreign datasets that the query engine attaches before the query runs.
Serializes to the .model file format and parses it back: a dialect: line, an optional
attach: JSON line, then the SQL body. dlt encrypts the secret-flagged statements in
that JSON line.
The generated file has no anti-tamper protection. dlt encrypts the text so that a backup of the pipeline working directory does not leak sensitive information.
attach
@property
def attach() -> Optional[List[TAttachInfo]]
Serializable attach info that attaches the foreign datasets again before the query runs.
Decrypts the secret-flagged statements of a model that comes from a file. Only a step
that runs these statements needs the encryption key.
with_query
def with_query(query: str, dialect: Optional[str] = None) -> "SqlModel"
Returns a copy of this model. The copy carries the new query and dialect.
The new model keeps the attach info in its stored form. This method therefore rewrites a model that comes from a file, and does not decrypt the secrets it carries.
Arguments:
querystr - The SQL query that replaces the query of this model.dialectOptional[str] - The dialect ofquery.
Returns:
SqlModel- A new model with the same attach info.
Raises:
ValueError- If the parsed query is not asqlglot.exp.Select.
__str__
def __str__() -> str
Serializes to .model text: a dialect line, an optional attach line, then the SQL body.
This method encrypts the secret-flagged attach statements with the active pipeline
encryption.
from_query_string
@classmethod
def from_query_string(
cls,
query: str,
dialect: Optional[str] = None,
attach: Optional[List[TAttachInfo]] = None) -> "SqlModel"
Creates a SqlModel from a raw SQL query string. The method raises when the query
is not a SELECT.
Arguments:
query- The raw SQL query string.dialect- The SQL dialect to use for parsing.attach- Serializable attach info that dlt carries to the load step.
Returns:
An instance of SqlModel with the normalized query and dialect.
Raises:
ValueError- If the parsed query is not asqlglot.exp.Select.
from_file
@classmethod
def from_file(
cls,
file_obj: IO[str],
fallback_dialect: Optional["TSqlGlotDialect"] = None) -> "SqlModel"
Creates a SqlModel from the .model text that str(model) writes.
Reads the dialect: line, and uses fallback_dialect when that line has no dialect.
Then reads an optional attach: line and keeps it in its stored form. The rest of the
file is the SQL body. This method parses neither the SQL nor the attach: line. It
decrypts the secrets of the attach: line only when a caller reads attach.
Arguments:
file_objIO[str] - A file-like object opened in text mode.fallback_dialectOptional[str] - The dialect to use when the first line has no dialect.
Returns:
An instance of SqlModel with the stored query, dialect and attach info.