dlt.extract.incremental
Incremental Objects
@configspec
class Incremental(ItemTransform[TDataItem, IncrementalCustomMetrics],
BaseConfiguration, Generic[TCursorValue])
Adds incremental extraction for a resource by storing a cursor value in persistent state.
The cursor could for example be a timestamp for when the record was created and you can use this to load only new records created since the last run of the pipeline.
To use this the resource function should have an argument either type annotated with Incremental or a default Incremental instance.
For example:
When the resource has a primary_key specified this is used to deduplicate overlapping items with the same cursor value.
Alternatively you can use this class as transform step and add it to any resource. For example:
@dlt.resource(primary_key='id')
def some_data(created_at=dlt.sources.incremental('created_at', '2023-01-01T00:00:00Z'):
yield from request_data(created_after=created_at.last_value)
@dlt.resource
def some_data():
last_value = dlt.sources.incremental.from_existing_state("some_data", "item.ts")
...
r = some_data().add_step(dlt.sources.incremental("item.ts", initial_value=now, primary_key="delta"))
info = p.run(r, destination="duckdb")
Arguments:
cursor_path- The name or a JSON path to a cursor field. Uses the same names of fields as in your JSON document, before they are normalized to store in the database.initial_value- Optional value used forlast_valuewhen no state is available, e.g. on the first run of the pipeline. If not providedlast_valuewill beNoneon the first run.last_value_func- Callable used to determine which cursor value to save in state. It is called with a list of the stored state value and all cursor vals from currently processing items. Default ismaxprimary_key- Optional primary key used to deduplicate data. If not provided, a primary key defined by the resource will be used. Pass a tuple to define a compound key. Pass empty tuple to disable unique checksend_value- Optional value used to load a limited range of records betweeninitial_valueandend_value. Use in conjunction withinitial_value, e.g. load records from given monthincremental(initial_value="2022-01-01T00:00:00Z", end_value="2022-02-01T00:00:00Z")Note, when this is set the incremental filtering is stateless andinitial_valuealways supersedes any previous incremental value in state.row_order- Declares that data source returns rows in descending (desc) or ascending (asc) order as defined bylast_value_func. If row order is know, Incremental class is able to stop requesting new rows by closing pipe generator. This prevents getting more data from the source. Defaults to None, which means that row order is not known.allow_external_schedulers- If set to True, allows dlt to look for external schedulers from which it will take "initial_value" and "end_value" resulting in loading only specified range of data. Currently Airflow scheduler is detected: "data_interval_start" and "data_interval_end" are taken from the context and passed Incremental class. The values passed explicitly to Incremental will be ignored. Note that if logical "end date" is present then also "end_value" will be set which means that resource state is not used and exactly this range of date will be loaded Defaults to None in which case the runtime may enable it viaTimeIntervalContext. Set to False to opt out unconditionally.on_cursor_value_missing- Specify what happens when the cursor_path does not exist in a record or a record hasNoneat the cursor_path: raise, include, excludelag- Optional value used to define a lag or attribution window. For datetime cursors, this is interpreted as seconds. For other types, it uses the + or - operator depending on the last_value_func.range_start- Decide whether the incremental filtering range isopenorclosedon the start value side. Default isclosed. Setting this toopenmeans that items with the same cursor value as the last value from the previous run (orinitial_value) are excluded from the result. Theopenrange disables deduplication logic so it can serve as an optimization when you know cursors don't overlap between pipeline runs.range_end- Decide whether the incremental filtering range isopenorclosedon the end value side. Default isopen(exactend_valueis excluded). Setting this toclosedmeans that items with the exact same cursor value as theend_valueare included in the result.
placement_affinity
stick to end
set_deduplication_key
def set_deduplication_key(value: Optional[TTableHintTemplate[TColumnNames]],
from_hints: bool) -> None
Set deduplication key. Manage priority with from_hints flag so values coming
from hints never overwrite values set directly
from_existing_state
@classmethod
def from_existing_state(cls, resource_name: str,
cursor_path: str) -> "Incremental[TCursorValue]"
Create Incremental instance from existing state.
merge
def merge(other: "Incremental[TCursorValue]") -> "Incremental[TCursorValue]"
Create a new incremental instance which merges the two instances.
Only properties which are not None from other override the current instance properties.
This supports use cases with partial overrides, such as:
def my_resource(updated=incremental('updated', initial_value='1970-01-01'))
...
my_resource(updated=incremental(initial_value='2023-01-01', end_value='2023-02-01'))
copy
def copy(with_transient_state: bool = False) -> "Incremental[TCursorValue]"
Creates a copy of this incremental with the same configuration.
The copy is never bound to a pipe. By default it also carries no transient state.
Arguments:
with_transient_statebool - WhenTrue, also copy the transient state (the cached state,start_valueand last value) so the copy resolves the same range. The copy owns its state and evolves it independently. Defaults toFalse.
Returns:
Incremental[TCursorValue]- The copied instance.
with_cursor
def with_cursor(cursor: str) -> "Incremental[TCursorValue]"
Creates a copy of this incremental that reads a different cursor.
The copy carries the runtime state (see copy), so it filters on cursor using the
same range and evolves its own state independently.
Arguments:
cursorstr - Cursor path (column name or JSONPath) for the copy.
Returns:
Incremental[TCursorValue]- The copy reading fromcursor.
advance
def advance(last_value: TCursorValue) -> Self
Pins the range end to last_value and passes rows through unfiltered.
Use when the range was already applied elsewhere, ie. pushed down to SQL. A None
value leaves the range as is and only disables filtering.
Arguments:
last_valueTCursorValue - The value to pin the range end to.
Returns:
Incremental[TCursorValue]- This instance, so calls can be chained.
is_unique_cursor
def is_unique_cursor() -> bool
True when the primary key is exactly the cursor column, so cursor values are unique.
cursor_value_hash
def cursor_value_hash(value: Any, legacy: bool = False) -> str
Dedup hash of a cursor value, legacy matches hashes written before 1.29
unique_boundary_consumed
def unique_boundary_consumed() -> bool
True when state marks the row at last_value as already loaded.
get_cursor_column_name
def get_cursor_column_name() -> Optional[str]
Return the name of the cursor column if the cursor path resolves to a single column
get_current_range
def get_current_range(
apply_lag: bool = True
) -> Tuple[Optional[TCursorValue], Optional[TCursorValue]]
Resolve current (start, end) cursor range based on internal incremental state
Arguments:
apply_lagbool - When True, the returnedstartis the lag-adjustedstart_valueset bybind(). When False, the rawstart_valuepersisted in cached state is returned.
Returns:
Tuple[Optional[TCursorValue], Optional[TCursorValue]]: (start, end) range.
get_state
def get_state() -> IncrementalColumnState
Returns or creates an Incremental state for a particular cursor column
If end_value is set, a mock state is created that will be discarded after extract step Otherwise state is taken from current pipeline and will be persisted in it
get_incremental_value_type
def get_incremental_value_type() -> Type[Any]
Infers the type of incremental value from a class of an instance if those preserve the Generic arguments information.
bind
def bind(pipe: SupportsPipe) -> "Incremental[TCursorValue]"
Called by pipe just before evaluation
can_close
def can_close() -> bool
Checks if incremental is out of range and can be closed.
Returns true only when row_order was set and
- results are ordered ascending and are above the range end (end_value)
- results are ordered descending and are below or equal the range start (start_value)
IncrementalResourceWrapper Objects
class IncrementalResourceWrapper(ItemTransform[TDataItem,
IncrementalCustomMetrics])
placement_affinity
stick to end
__init__
def __init__() -> None
Creates a wrapper over a resource function that accepts Incremental instance in its argument to perform incremental loading.
The wrapper delays instantiation of the Incremental to the moment of actual execution and is currently used by dlt.resource decorator.
The wrapper explicitly (via resource_name) parameter binds the Incremental state to a resource state.
Note that wrapper implements FilterItem transform interface and functions as a processing step in the before-mentioned resource pipe.
set_deduplication_key
def set_deduplication_key(value: Optional[TTableHintTemplate[TColumnNames]],
from_hints: Literal[True]) -> None
Set deduplication key, wrapper is always called from resource so from_hints must be always True
inject_implicit_incremental_arg
@staticmethod
def inject_implicit_incremental_arg(
incremental: Optional[Union[Incremental[Any],
"IncrementalResourceWrapper"]],
sig: inspect.Signature,
func_args: Tuple[Any],
func_kwargs: Dict[str, Any],
fallback: Optional[Incremental[Any]] = None
) -> Tuple[Tuple[Any], Dict[str, Any], Optional[Incremental[Any]]]
Inject the incremental instance into function arguments if the function has an incremental argument without default in its signature and it is not already set in the arguments.
Returns:
Tuple of the new args, kwargs and the incremental instance that was injected (if any)
wrap
def wrap(sig: inspect.Signature, func: TFun) -> TFun
Wrap the callable to inject an Incremental object configured for the resource.
set_incremental
def set_incremental(incremental: Optional[TIncrementalConfig],
from_hints: bool = False) -> None
Sets the incremental. If incremental was set from_hints, it can only be changed in the same manner
allow_external_schedulers
@property
def allow_external_schedulers() -> bool
Allows the Incremental instance to get its initial and end values from external schedulers like Airflow
custom_metrics
@property
def custom_metrics() -> IncrementalCustomMetrics
Returns custom metrics of the Incremental object itself if exists