singer_sdk.Stream

class singer_sdk.Stream(tap: singer_sdk.plugin_base.PluginBase, schema: Optional[Union[str, os.PathLike, Dict[str, Any], singer.schema.Schema]] = None, name: Optional[str] = None)

Abstract base class for tap streams.

apply_catalog(catalog: singer_sdk.helpers._singer.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 config

Get stream configuration.

Returns

A frozen (read-only) config dictionary map.

property descendent_streams

Get child streams.

Returns

A list of all children, recursively.

finalize_state_progress_markers(state: Optional[dict] = 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_child_context(record: dict, context: Optional[dict]) → dict

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.

Parameters

• record – Individual record in the stream.

• context – Stream partition or context dictionary.

Returns

A dictionary with context values for a child stream.

Raises

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

get_context_state(context: Optional[dict]) → 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: Optional[dict]) → Iterable[Union[dict, Tuple[dict, dict]]]

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

Each row 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. More info: Parent-Child Streams

Parameters

context – Stream partition or context dictionary.

get_replication_key_signpost(context: Optional[dict]) → Optional[Union[datetime.datetime, Any]]

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: SDK Implementation Details - 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: Optional[dict]) → Optional[Any]

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: Optional[dict]) → Optional[datetime.datetime]

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

Check descendents.

Returns

True if any child streams are selected, recursively.

property is_sorted

Check if stream is sorted.

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

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

Returns

True if stream is sorted. Defaults to False.

property is_timestamp_replication_key

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.

property mask

Get a boolean mask for stream and property selection.

Returns

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

property metadata

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.

property partitions

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: Optional[dict] = None) → Optional[dict]

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

Get primary keys.

Returns

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

property replication_key

Get replication key.

Returns

Replication key for the stream.

property replication_method

Get replication method.

Returns

Replication method to be used for this stream.

reset_state_progress_markers(state: Optional[dict] = 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

Get schema.

Returns

JSON Schema dictionary for this stream.

property schema_filepath

Get path to schema file.

Returns

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

property selected

Check isf stream is selected.

Returns

True if the stream is selected.

property state_partitioning_keys

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

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

Get writable state.

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.

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

Returns

A writable state dict for this stream.

sync(context: Optional[dict] = 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.

property tap_state

Return a writeable state dict for the entire tap.

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

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.

Returns

A writeable state dict for the entire tap.

property tap_stream_id

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.