class singer_sdk.InlineMapper#

Bases: PluginBase, SingerReader, SingerWriter

Abstract base class for inline mappers.

__init__(*, config: dict | PurePath | str | list[PurePath | str] | None = None, parse_env_config: bool = False, validate_config: bool = True) None#

Create the tap or target.

  • config – May be one or more paths, either as str or PurePath objects, or it can be a predetermined config dict.

  • parse_env_config – True to parse settings from env vars.

  • validate_config – True to require validation of config settings.


ValueError – If config is not a dict or path string.

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.


config_jsonschema – [description]

classmethod cb_version(ctx: Context, param: Option, value: bool) None#

CLI callback to print the plugin version and exit.

  • ctx – Click context.

  • param – Click parameter.

  • value – Boolean indicating whether to print the version.

property config: Mapping[str, Any]#

Get config.


A frozen (read-only) config dictionary map.

static config_from_cli_args(*args: str) tuple[list[Path], bool]#

Parse CLI arguments into a config dictionary.


args – CLI arguments.


FileNotFoundError – If the config file does not exist.


A tuple containing the config dictionary and a boolean indicating whether the config file was found.

deserialize_json(line: str) dict#

Deserialize a line of json.


line – A single line of json.


A dictionary of the deserialized json.


json.decoder.JSONDecodeError – raised if any lines are not valid json

format_message(message: Message) str#

Format a message as a JSON string.


message – The message to format.


The formatted message.

classmethod get_plugin_version() str#

Return the package version number.


The package version number.

classmethod get_sdk_version() str#

Return the package version number.


The package version number.

classmethod get_singer_command() Command#

Execute standard CLI handler for inline mappers.


A click.Command object.

classmethod get_supported_python_versions() list[str] | None#

Return the supported Python versions.


A list of supported Python versions.

property initialized_at: int#

Start time of the plugin.


The start time of the plugin.

classmethod invoke(*, about: bool = False, about_format: str | None = None, config: tuple[str, ...] = (), file_input: IO[str] | None = None) None#

Invoke the mapper.

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

  • file_input – Optional file to read input from.

listen(file_input: IO[str] | None = None) None#

Read from input until all messages are processed.


file_input – Readable stream of messages. Defaults to standard in.

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

abstract map_activate_version_message(message_dict: dict) t.Iterable[singer.Message]#

Map a version message to zero or more new messages.


message_dict – An ACTIVATE_VERSION message JSON dictionary.

map_batch_message(message_dict: dict) t.Iterable[singer.Message]#

Map a batch message to zero or more new messages.


message_dict – A BATCH message JSON dictionary.


NotImplementedError – if not implemented by subclass.

abstract map_record_message(message_dict: dict) t.Iterable[singer.Message]#

Map a record message to zero or more new messages.


message_dict – A RECORD message JSON dictionary.

abstract map_schema_message(message_dict: dict) t.Iterable[singer.Message]#

Map a schema message to zero or more new messages.


message_dict – A SCHEMA message JSON dictionary.

abstract map_state_message(message_dict: dict) t.Iterable[singer.Message]#

Map a state message to zero or more new messages.


message_dict – A STATE message JSON dictionary.

property mapper: PluginMapper#

Plugin mapper for this tap.


A PluginMapper object.


MapperNotInitialized – If the mapper has not been initialized.

name: str#

The executable name of the tap or target plugin. e.g. tap-foo

package_name: str | None = None#

The package name of the plugin. e.g meltanolabs-tap-foo

classmethod print_about(output_format: str | None = None) None#

Print capabilities and other tap metadata.


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.


print_fn – A function to use to display the plugin version. Defaults to print.

setup_mapper() None#

Initialize the plugin mapper for this tap.

property state: dict#

Get state.


NotImplementedError – If the derived plugin doesn’t override this method.

write_message(message: Message) None#

Write a message to stdout.


message – The message to write.