singer_sdk.Stream#

class singer_sdk.Stream#

Abstract base class for tap streams.

ABORT_AT_RECORD_COUNT: int | None = None#: If set, raise MaxRecordsLimitException if the limit is exceeded.

STATE_MSG_FREQUENCY = 10000#: Number of records between state messages.

TYPE_CONFORMANCE_LEVEL = 1#

Type conformance level for this stream.

Field types in the schema are used to convert record field values to the correct type.

Available options are:

TypeConformanceLevel.NONE: No conformance is performed.
TypeConformanceLevel.RECURSIVE: Conformance is performed recursively through all nested levels in the record.
TypeConformanceLevel.ROOT_ONLY: Conformance is performed only on the root level.

Init tap stream.

Parameters:

tap – Singer Tap this stream belongs to.
schema – JSON schema for records in this stream.
name – Name of this stream.

Raises:

ValueError – TODO
FileNotFoundError – TODO

apply_catalog(catalog: Catalog) → None#

Apply a catalog dict, updating any settings overridden within the catalog.

Developers may override this method in order to introduce advanced catalog parsing, or to explicitly fail on advanced catalog customizations which are not supported by the tap.

Parameters:: catalog – Catalog object passed to the tap. Defines schema, primary and replication keys, as well as selection metadata.

property check_sorted: bool#

Check if stream is sorted.

This setting enables additional checks which may trigger InvalidStreamSortException if records are found which are unsorted.

Returns:: True if sorting is checked. Defaults to True.

compare_start_date(value: str, start_date_value: str) → str#

Compare a bookmark value to a start date and return the most recent value.

If the replication key is a datetime-formatted string, this method will parse the value and compare it to the start date. Otherwise, the bookmark value is returned.

If the tap uses a non-datetime replication key (e.g. an UNIX timestamp), the developer is encouraged to override this method to provide custom logic for comparing the bookmark value to the start date.

Parameters:

value – The replication key value.
start_date_value – The start date value from the config.

Returns:

The most recent value between the bookmark and start date.

property config: Mapping[str, Any]#

Get stream configuration.

Returns:: A frozen (read-only) config dictionary map.

property descendent_streams: list[singer_sdk.streams.core.Stream]#

Get child streams.

Returns:: A list of all children, recursively.

finalize_state_progress_markers(state: dict | None = None) → None#

Reset progress markers. If all=True, all state contexts will be finalized.

This method is internal to the SDK and should not need to be overridden.

If all=True and the stream has children, child streams will also be finalized.

Parameters:: state – State object to promote progress markers with.

get_batch_config(config: Mapping) → singer_sdk.helpers._batch.BatchConfig | None#

Return the batch config for this stream.

Parameters:: config – Tap configuration dictionary.
Returns:: Batch config for this stream.

get_batches(batch_config: BatchConfig, context: dict | None = None) → Iterable[tuple[singer_sdk.helpers._batch.BaseBatchFileEncoding, list[str]]]#

Batch generator function.

Developers are encouraged to override this method to customize batching behavior for databases, bulk APIs, etc.

Parameters:

batch_config – Batch config for this stream.
context – Stream partition or context dictionary.

Yields:

A tuple of (encoding, manifest) for each batch.

get_child_context(record: dict, context: dict | None) → dict | None#

Return a child context object from the record and optional provided context.

By default, will return context if provided and otherwise the record dict.

Developers may override this behavior to send specific information to child streams for context.

Return None if no child streams should be synced, for example if the parent record was deleted and the child records can no longer be synced.

Parameters:

record – Individual record in the stream.
context – Stream partition or context dictionary.

Returns:

A dictionary with context values for a child stream, or None if no child streams should be synced.

Raises:

NotImplementedError – If the stream has children but this method is not overriden.

get_context_state(context: dict | None) → dict#

Return a writable state dict for the given context.

Gives a partitioned context state if applicable; else returns stream state. A blank state will be created in none exists.

This method is internal to the SDK and should not need to be overridden. Developers may access this property but this is not recommended except in advanced use cases. Instead, developers should access the latest stream replication key values using get_starting_timestamp() for timestamp keys, or get_starting_replication_key_value() for non-timestamp keys.

Partition level may be overridden by state_partitioning_keys if set.

Parameters:: context – Stream partition or context dictionary.
Returns:: A partitioned context state if applicable; else returns stream state. A blank state will be created in none exists.

abstract get_records(context: dict | None) → Iterable[dict | tuple[dict, dict | None]]#

Abstract record generator function. Must be overridden by the child class.

Each record emitted should be a dictionary of property names to their values. Returns either a record dict or a tuple: (record_dict, child_context)

A method which should retrieve data from the source and return records incrementally using the python yield operator.

Only custom stream types need to define this method. REST and GraphQL streams should instead use the class-specific methods for REST or GraphQL, respectively.

This method takes an optional context argument, which can be safely ignored unless the stream is a child stream or requires partitioning. More info: Stream Partitioning.

Parent streams can optionally return a tuple, in which case the second item in the tuple being a child_context dictionary for the stream’s context.

If the child context object in the tuple is None, the child streams will be skipped. This is useful for cases where the parent record was deleted and the child records can no longer be synced.

More info: Parent-Child Streams

Parameters:: context – Stream partition or context dictionary.

get_replication_key_signpost(context: dict | None) → datetime.datetime | Any | None#

Get the replication signpost.

For timestamp-based replication keys, this defaults to utc_now(). For non-timestamp replication keys, default to None. For consistency in subsequent calls, the value will be frozen (cached) at its initially called state, per partition argument if applicable.

Developers may optionally override this method in advanced use cases such as unsorted incremental streams or complex hierarchical stream scenarios. For more info: Stream State

Parameters:: context – Stream partition or context dictionary.
Returns:: Max allowable bookmark value for this stream’s replication key.

get_starting_replication_key_value(context: dict | None) → Any | None#

Get starting replication key.

Will return the value of the stream’s replication key when –state is passed. If no prior state exists, will return None.

Developers should use this method to seed incremental processing for non-datetime replication keys. For datetime and date replication keys, use get_starting_timestamp()

Parameters:: context – Stream partition or context dictionary.
Returns:: Starting replication value.

get_starting_timestamp(context: dict | None) → datetime.datetime | None#

Get starting replication timestamp.

Will return the value of the stream’s replication key when –state is passed. If no state exists, will return start_date if set, or None if neither the stream state nor start_date is set.

Developers should use this method to seed incremental processing for date and datetime replication keys. For non-datetime replication keys, use get_starting_replication_key_value()

Parameters:: context – Stream partition or context dictionary.
Returns:: start_date from config, or state value if using timestamp replication.
Raises:: ValueError – If the replication value is not a valid timestamp.

property has_selected_descendents: bool#

Check descendents.

Returns:: True if any child streams are selected, recursively.

property is_sorted: bool#

Expect stream to be sorted.

When True, incremental streams will attempt to resume if unexpectedly interrupted.

Returns:: True if stream is sorted. Defaults to False.

property is_timestamp_replication_key: bool#

Check is replication key is a timestamp.

Developers can override to True in order to force this value, although this should not be required in most use cases since the type can generally be accurately detected from the JSON Schema.

Returns:: True if the stream uses a timestamp-based replication key.

log_sync_costs() → None#

Log a summary of Sync costs.

The costs are calculated via calculate_sync_cost. This method can be overridden to log results in a custom format. It is only called once at the end of the life of the stream.

property mask: SelectionMask#

Get a boolean mask for stream and property selection.

Returns:: A mapping of breadcrumbs to boolean values, representing stream and field selection.

property metadata: MetadataMapping#

Get stream metadata.

Metadata attributes (inclusion, selected, etc.) are part of the Singer spec.

Metadata from an input catalog will override standard metadata.

Returns:: A mapping from property breadcrumbs to metadata objects.

parent_stream_type: type[Stream] | None = None#: Parent stream type for this stream. If this stream is a child stream, this should be set to the parent stream class.

property partitions: list[dict] | None#

Get stream partitions.

Developers may override this property to provide a default partitions list.

By default, this method returns a list of any partitions which are already defined in state, otherwise None.

Returns:: A list of partition key dicts (if applicable), otherwise None.

post_process(row: dict, context: dict | None = None) → dict | None#

As needed, append or transform raw data to match expected structure.

Optional. This method gives developers an opportunity to “clean up” the results prior to returning records to the downstream tap - for instance: cleaning, renaming, or appending properties to the raw record result returned from the API.

Developers may also return None from this method to filter out invalid or not-applicable records from the stream.

Parameters:

row – Individual record in the stream.
context – Stream partition or context dictionary.

Returns:

The resulting record dict, or None if the record should be excluded.

property primary_keys: list[str] | None#

Get primary keys.

Returns:: A list of primary key(s) for the stream.

property replication_key: str | None#

Get replication key.

Returns:: Replication key for the stream.

property replication_method: str#

Get replication method.

Returns:: Replication method to be used for this stream.

reset_state_progress_markers(state: dict | None = None) → None#

Reset progress markers. If all=True, all state contexts will be set.

This method is internal to the SDK and should not need to be overridden.

Parameters:: state – State object to promote progress markers with.

property schema: dict#

Get schema.

Returns:: JSON Schema dictionary for this stream.

property schema_filepath: pathlib.Path | None#

Get path to schema file.

Returns:: Path to a schema file for the stream or None if n/a.

property selected: bool#

Check if stream is selected.

Returns:: True if the stream is selected.

selected_by_default: bool = True#: Whether this stream is selected by default in the catalog.

property state_partitioning_keys: list[str] | None#

Get state partition keys.

If not set, a default partitioning will be inherited from the stream’s context. If an empty list is set ([]), state will be held in one bookmark per stream.

Returns:: Partition keys for the stream state bookmarks.

property stream_maps: list[singer_sdk.mapper.StreamMap]#

Get stream transformation maps.

The 0th item is the primary stream map. List should not be empty.

Returns:: A list of one or more map transformations for this stream.

property stream_state: dict#

Get writable state.

A blank state entry will be created if one doesn’t already exist.

Returns:: A writable state dict for this stream.

sync(context: dict | None = None) → None#

Sync this stream.

This method is internal to the SDK and should not need to be overridden.

Parameters:: context – Stream partition or context dictionary.
Raises:: Exception – Any exception raised by the sync process.

property tap_state: dict#

Return a writeable state dict for the entire tap.

Note: This dictionary is shared (and writable) across all streams.

Returns:: A writeable state dict for the entire tap.

property tap_stream_id: str#

Return a unique stream ID.

Default implementations will return self.name but this behavior may be overridden if required by the developer.

Returns:: Unique stream ID.