class singer_sdk.Tap

Abstract base class for taps.

The Tap class governs configuration, validation, and stream discovery for tap plugins.

__init__(config: Optional[Union[dict, PurePath, str, List[Union[PurePath, str]]]] = None, catalog: Optional[Union[PurePath, str, dict, Catalog]] = None, state: Optional[Union[PurePath, str, dict]] = None, parse_env_config: bool = False, validate_config: bool = True) None

Initialize the tap.

  • config – Tap configuration. Can be a dictionary, a single path to a configuration file, or a list of paths to multiple configuration files.

  • catalog – Tap catalog. Can be a dictionary or a path to the catalog file.

  • state – Tap state. Can be dictionary or a path to the state file.

  • parse_env_config – Whether to look for configuration values in environment variables.

  • validate_config – True to require validation of config settings.

property catalog: Catalog

Get the tap’s working catalog.


A Singer catalog object.

property catalog_dict: dict

Get catalog dictionary.


The tap’s catalog as a dict

property catalog_json_text: str

Get catalog JSON.


The tap’s catalog as formatted JSON text.

cli = <Command cli>
discover_streams() List[Stream]

Initialize all available streams and return them as a list.


List of discovered Stream objects.


NotImplementedError – If the tap implementation does not override this method.

property input_catalog: Optional[Catalog]

Get the catalog passed to the tap.


Catalog dictionary input, or None if not provided.

load_state(state: Dict[str, Any]) None

Merge or initialize stream state with the provided state dictionary input.

Override this method to perform validation and backwards-compatibility patches on self.state. If overriding, we recommend first running super().load_state(state) to ensure compatibility with the SDK.


state – Initialize the tap’s state with this value.


ValueError – If the tap’s own state is None, meaning it has not been initialized.

load_streams() List[Stream]

Load streams from discovery and initialize DAG.

Return the output of self.discover_streams() to enumerate discovered streams.


A list of discovered streams, ordered by name.

run_connection_test() bool

Run connection test.


True if the test succeeded.

run_discovery() str

Write the catalog json to STDOUT and return as a string.


The catalog as a string of JSON.

property state: dict

Get tap state.


The tap’s state dictionary


RuntimeError – If state has not been initialized.

property streams: Dict[str, Stream]

Get streams discovered or catalogued for this tap.

Results will be cached after first execution.


A mapping of names to streams, using discovery or a provided catalog.

sync_all() None

Sync all streams.

write_schemas() None

Write a SCHEMA message for all known streams to STDOUT.