singer_sdk.Stream

class singer_sdk.Stream

Abstract base class for tap streams.

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: Conformance is performed only on the root level.

__init__(tap: TapBaseClass, schema: str | PathLike | dict[str, Any] | singer.Schema | None = None, name: str | None = None) → None

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.

batch_size: int = 1000

Max number of records to write to each batch file.

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[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) → 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[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

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: 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]]

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. 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: 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: 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: 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.

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[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.

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: 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.

property tap_state: dict

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: 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.