dlt.common.storages.load_package
TJobFileFormat
Loader file formats with internal job types
TPipelineStateDoc Objects
class TPipelineStateDoc(TypedDict)
Corresponds to the StateInfo Tuple
TLoadPackageDropTablesState Objects
class TLoadPackageDropTablesState(TypedDict)
refresh
Refresh mode that generated the drop/truncate lists below
dropped_tables
List of tables that are to be dropped from the schema and destination (i.e. when refresh mode is used)
truncated_tables
List of tables that are to be truncated in the destination (i.e. when refresh='drop_data' mode is used)
TLoadPackageState Objects
class TLoadPackageState(TVersionedState, TLoadPackageDropTablesState)
created_at
Timestamp when the load package was created, in UTC
pipeline_state
Pipeline state, added at the end of the extraction phase
destination_state
private space for destinations to store state relevant only to the load package
placeholders
Placeholder values resolved for this package, by table name
load_metrics
Per-job load metrics, persisted so they survive process restarts
applied_dropped_tables
Names of dropped_tables actually dropped at the destination by the load step
applied_truncated_tables
Names of truncated_tables actually truncated at the destination by the load step
abort_requested
Flag indicating the package should be aborted on next load
TLoadPackage Objects
class TLoadPackage(TypedDict)
load_id
Load id
state
State of the load package
create_load_id
def create_load_id() -> str
Creates new package load id which is the current unix timestamp converted to string. Load ids must have the following properties:
- They must maintain increase order over time for a particular dlt schema loaded to particular destination and dataset
dltexecutes packages in order of load idsdltconsiders a state with the highest load id to be the most up to date when restoring state from destination
ParsedLoadJobFileName Objects
class ParsedLoadJobFileName(NamedTuple)
Represents a file name of a job in load package. The file name contains name of a table, number of times the job was retried, extension and a 5 bytes random string to make job file name unique. The job id does not contain retry count and is immutable during loading of the data
job_id
def job_id() -> str
Unique identifier of the job
file_name
def file_name() -> str
A name of the file with the data to be loaded
with_retry
def with_retry() -> "ParsedLoadJobFileName"
Returns a job with increased retry count
to_reference_file_name
def to_reference_file_name() -> str
Returns a file name for a reference job
to_exception_file_name
def to_exception_file_name() -> str
Returns a file name for an exception
full_extension
def full_extension() -> str
Returns the full file extension
group_jobs_by_table_name
def group_jobs_by_table_name(
jobs: Iterable[ParsedLoadJobFileName]
) -> dict[str, list[ParsedLoadJobFileName]]
Returns dictionary with table names as keys and list of jobs for those tables as values.
_LoadPackageInfo Objects
class _LoadPackageInfo(NamedTuple)
refresh
Refresh mode the package was extracted with
dropped_tables
Names of tables dropped in the destination
truncated_tables
Names of tables truncated in the destination
has_pending_transitions
True when jobs committed to the destination but were not yet moved (crash mid-load)
LoadPackageInfo Objects
class LoadPackageInfo(SupportsHumanize, _LoadPackageInfo)
schema_migrated
@property
def schema_migrated() -> bool
True when the schema was applied (migrated) at the destination for this package.
PackageStorage Objects
class PackageStorage()
APPLIED_SCHEMA_UPDATES_FILE_NAME
updates applied to the destination
__init__
def __init__(storage: FileStorage, initial_state: TLoadPackageStatus) -> None
Creates storage that manages load packages with root at storage and initial package state initial_state
get_package_path
def get_package_path(load_id: str) -> str
Gets path of the package relative to storage root
get_job_state_folder_path
def get_job_state_folder_path(load_id: str, state: TPackageJobState) -> str
Gets path to the jobs in state in package load_id, relative to the storage root
get_job_file_path
def get_job_file_path(load_id: str, state: TPackageJobState,
file_name: str) -> str
Get path to job with file_name in state in package load_id, relative to the storage root
list_packages
def list_packages() -> Sequence[str]
Lists all load ids in storage, earliest first
NOTE: Load ids are sorted alphabetically. This class does not store package creation time separately.
list_retried_new_jobs
def list_retried_new_jobs(
load_id: str) -> Sequence[Tuple[str, TPackageJobState]]
Lists (job file, folder) tuples of jobs pending a retry. Reads pending transitions but does not resolve them.
list_failed_jobs_infos
def list_failed_jobs_infos(load_id: str) -> Sequence[LoadJobInfo]
List all failed jobs and associated error messages for a load package with load_id
import_job
def import_job(load_id: str,
job_file_path: str,
job_state: TPackageJobState = "new_jobs") -> None
Adds new job by moving the job_file_path into new_jobs of package load_id
fail_retried_job
def fail_retried_job(load_id: str, file_name: str) -> str
Fails a retried job that is currently in new_jobs. Copies the last exception to failed_jobs/; the full per-retry history is kept in the exceptions folder.
retry_failed_job
def retry_failed_job(load_id: str, file_name: str) -> str
Retry a job that is currently in failed_jobs. Moves exception to exceptions/ folder and job to new_jobs with retry count increased.
save_pending_transition
def save_pending_transition(load_id: str,
file_name: str,
state: str,
failed_message: Optional[str] = None) -> None
Atomically saves a pending transition marker with state and optional error.
load_pending_transition
def load_pending_transition(
load_id: str, file_name: str) -> Optional[Tuple[str, Optional[str]]]
Returns (state, failed_message) if a pending transition exists.
list_pending_transitions
def list_pending_transitions(load_id: str) -> Sequence[str]
Lists file names in the pending transitions folder.
clear_pending_transition
def clear_pending_transition(load_id: str, file_name: str) -> None
Removes a pending transition marker file.
migrate_package
def migrate_package(load_id: str, from_version: semver.Version,
to_version: semver.Version) -> None
Migrates load package using load/normalize storage version
complete_loading_package
def complete_loading_package(load_id: str,
load_state: TLoadPackageStatus) -> str
Completes loading the package by writing marker file with`package_state. Returns path to the completed package
remove_completed_jobs
def remove_completed_jobs(load_id: str) -> None
Deletes completed jobs. If package has failed jobs, nothing gets deleted.
get_schema_update_file
def get_schema_update_file(load_id: str) -> Optional[TSchemaTables]
Reads the update file from load package load_id and returns its content.
Returns none if update file is already processed
schema_name
def schema_name(load_id: str) -> str
Gets schema name associated with the package
cancel
def cancel(load_id: str) -> None
Sets cancel flag currently used for inter-process signalling
raise_if_cancelled
def raise_if_cancelled(load_id: str) -> None
Raise an exception if package is cancelled
get_progress_dir
def get_progress_dir(load_id: str) -> str
Gets the relative path to the progress directory within a package.
create_progress_dir
def create_progress_dir(load_id: str) -> None
Creates the progress directory within a package.
write_progress
def write_progress(load_id: str, table_counts: Dict[str, int]) -> None
Writes buffered table→count pairs as a JSON progress file (atomic via rename).
collect_progress
def collect_progress(load_id: str) -> Dict[str, int]
Reads and deletes all progress files, returning aggregated table→items_count.
cleanup_progress
def cleanup_progress(load_id: str) -> None
Removes the progress directory and all its contents.
set_abort_flag
def set_abort_flag(load_id: str) -> None
Mark a package to be aborted on the next load.
Arguments:
load_id- Load package ID to mark for abort
has_abort_flag
def has_abort_flag(load_id: str) -> bool
Check if a package is marked for abort.
Arguments:
load_id- Load package ID to check
Returns:
True if the package is marked for abort
save_pipeline_state
def save_pipeline_state(load_id: str, state_blob: Optional[str],
schema_blobs: Dict[str, str]) -> bool
Saves local pipeline state and schema blobs taken at package start to .restore.
No-op when a snapshot already exists. Returns True when the snapshot was written.
load_pipeline_state
def load_pipeline_state(load_id: str) -> Tuple[Optional[str], Dict[str, str]]
Loads (state blob, schema blobs by name) from .restore of a package.
Raises:
FileNotFoundError- If the package has no completed snapshot.
abort_package
def abort_package(load_id: str) -> None
Aborts a package by moving all retried jobs in new_jobs to failed_jobs. Raises
LoadPackageInconsistent when jobs are still in started_jobs. Does NOT complete
the package - that's the loader's responsibility.
get_load_package_jobs
def get_load_package_jobs(
load_id: str) -> Dict[TPackageJobState, List[ParsedLoadJobFileName]]
Gets all jobs in a package and returns them as lists assigned to a particular state.
get_load_package_info
def get_load_package_info(load_id: str) -> LoadPackageInfo
Gets information on normalized/completed package with given load_id, all jobs and their statuses.
Will reach to the file system to get additional stats, mtime, also collects exceptions for failed jobs. NOTE: do not call this function often. it should be used only to generate metrics
get_job_failed_message
def get_job_failed_message(load_id: str, job: ParsedLoadJobFileName) -> str
Get exception message of a failed job.
get_last_job_exception
def get_last_job_exception(
load_id: str,
job: ParsedLoadJobFileName) -> Tuple[TExceptionType, Optional[str]]
Gets exception type and message saved at the last retry of a job in new_jobs.
Exception type defaults to transient when unknown - such job may be retried.
list_job_exceptions
def list_job_exceptions(
load_id: str,
job: ParsedLoadJobFileName) -> List[Tuple[int, TExceptionType, str]]
Reads all exception messages recorded for a job across its retries, oldest first, as (retry number, exception type, message) tuples.
job_to_job_info
def job_to_job_info(load_id: str, state: TPackageJobState,
job: ParsedLoadJobFileName) -> LoadJobInfo
Creates partial job info by converting job object. size, mtime and failed message will not be populated
is_package_partially_loaded
@staticmethod
def is_package_partially_loaded(package_info: LoadPackageInfo) -> bool
Checks if a package may have modified the destination: some jobs completed while others did not, the schema was migrated at the destination while jobs remained, a pending new job was retried more than once (an earlier attempt may have written), or a job committed but was not yet moved.
partially_loaded_warning
@staticmethod
def partially_loaded_warning(load_id: str) -> str
Warning shown for a package that may have modified the destination inconsistently.
load_package_state
def load_package_state() -> TLoadPackage
Get full load package state present in current context. Across all threads this will be the same in memory dict.
commit_load_package_state
def commit_load_package_state() -> None
Commit load package state present in current context. This is thread safe.
destination_state
def destination_state() -> DictStrAny
Get segment of load package state that is specific to the current destination.
clear_destination_state
def clear_destination_state(commit: bool = True) -> None
Clear segment of load package state that is specific to the current destination. Optionally commit to load package.