singer_sdk.Tap#
- class singer_sdk.Tap#
Bases:
PluginBase
,SingerWriter
Abstract base class for taps.
The Tap class governs configuration, validation, and stream discovery for tap plugins.
- __init__(*, config: dict | PurePath | str | list[PurePath | str] | None = None, catalog: PurePath | str | dict | Catalog | None = None, state: PurePath | str | dict | None = None, parse_env_config: bool = False, validate_config: bool = True, setup_mapper: bool = True) None #
Initialize the tap.
- Parameters:
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.
setup_mapper – True to initialize the plugin mapper.
- classmethod append_builtin_config(config_jsonschema: dict) None #
Appends built-in config to config_jsonschema if not already set.
To customize or disable this behavior, developers may either override this class method or override the capabilities property to disabled any unwanted built-in capabilities.
For all except very advanced use cases, we recommend leaving these implementations “as-is”, since this provides the most choice to users and is the most “future proof” in terms of taking advantage of built-in capabilities which may be added in the future.
- Parameters:
config_jsonschema – [description]
- classmethod cb_discover(ctx: Context, param: Option, value: bool) None #
CLI callback to run the tap in discovery mode.
- Parameters:
ctx – Click context.
param – Click option.
value – Whether to run in discovery mode.
- classmethod cb_test(ctx: Context, param: Option, value: bool) None #
CLI callback to run the tap in test mode.
- Parameters:
ctx – Click context.
param – Click option.
value – Whether to run in test mode.
- classmethod cb_version(ctx: Context, param: Option, value: bool) None #
CLI callback to print the plugin version and exit.
- Parameters:
ctx – Click context.
param – Click parameter.
value – Boolean indicating whether to print the version.
- static config_from_cli_args(*args: str) tuple[list[Path], bool] #
Parse CLI arguments into a config dictionary.
- Parameters:
args – CLI arguments.
- Raises:
FileNotFoundError – If the config file does not exist.
- Returns:
A tuple containing the config dictionary and a boolean indicating whether the config file was found.
- discover_streams() t.Sequence[Stream] #
Initialize all available streams and return them as a list.
- Returns:
List of discovered Stream objects.
- Raises:
NotImplementedError – If the tap implementation does not override this method.
- format_message(message: Message) str #
Format a message as a JSON string.
- Parameters:
message – The message to format.
- Returns:
The formatted message.
- classmethod get_plugin_version() str #
Return the package version number.
- Returns:
The package version number.
- classmethod get_sdk_version() str #
Return the package version number.
- Returns:
The package version number.
- classmethod get_singer_command() Command #
Execute standard CLI handler for taps.
- Returns:
A click.Command object.
- classmethod get_supported_python_versions() list[str] | None #
Return the supported Python versions.
- Returns:
A list of supported Python versions.
- classmethod invoke(*, about: bool = False, about_format: str | None = None, config: tuple[str, ...] = (), state: str | None = None, catalog: str | None = None) None #
Invoke the tap’s command line interface.
- Parameters:
about – Display package metadata and settings.
about_format – Specify output style for –about.
config – Configuration file location or ‘ENV’ to use environment variables. Accepts multiple inputs as a tuple.
catalog – Use a Singer catalog file with the tap.”,
state – Use a bookmarks file for incremental replication.
- 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.
- Parameters:
state – Initialize the tap’s state with this value.
- Raises:
ValueError – If the tap’s own state is None, meaning it has not been initialized.
- final load_streams() list[Stream] #
Load streams from discovery and initialize DAG.
Return the output of self.discover_streams() to enumerate discovered streams.
- Returns:
A list of discovered streams, ordered by name.
- classmethod print_about(output_format: str | None = None) None #
Print capabilities and other tap metadata.
- Parameters:
output_format – Render option for the plugin information.
- classmethod print_version(print_fn: ~typing.Callable[[~typing.Any], None] = <built-in function print>) None #
Print help text for the tap.
- Parameters:
print_fn – A function to use to display the plugin version. Defaults to
print()
.
- final run_connection_test() bool #
Run connection test, aborting each stream after 1 record.
- Returns:
True if the test succeeded.
- run_discovery() str #
Write the catalog json to STDOUT and return as a string.
- Returns:
The catalog as a string of JSON.
- final run_sync_dry_run(dry_run_record_limit: int | None = 1, streams: t.Iterable[Stream] | None = None) bool #
Run connection test.
Exceptions of type MaxRecordsLimitException and PartialSyncSuccessException will be ignored.
- Parameters:
dry_run_record_limit – The max number of records to sync per stream object.
streams – The streams to test. If omitted, all streams will be tested.
- Returns:
True if the test succeeded.
- write_message(message: Message) None #
Write a message to stdout.
- Parameters:
message – The message to write.
- property catalog: Catalog#
Get the tap’s working catalog.
- Returns:
A Singer catalog object.
- property catalog_json_text: str#
Get catalog JSON.
- Returns:
The tap’s catalog as formatted JSON text.
- property config: Mapping[str, Any]#
Get config.
- Returns:
A frozen (read-only) config dictionary map.
- dynamic_catalog: bool = False#
Whether the tap’s catalog is dynamic. Set to True if the catalog is generated dynamically (e.g. by querying a database’s system tables).
- property input_catalog: Catalog | None#
Get the catalog passed to the tap.
- Returns:
Catalog dictionary input, or None if not provided.
- property mapper: PluginMapper#
Plugin mapper for this tap.
- Returns:
A PluginMapper object.
- Raises:
MapperNotInitialized – If the mapper has not been initialized.
- property state: dict#
Get tap state.
- Returns:
The tap’s state dictionary
- Raises:
RuntimeError – If state has not been initialized.