datamill-co / singer-runner

A CLI and library to run Singer Taps and Targets

singer-runner

Singer Runner manages tap and target processes, as well as metrics, state, and configuration.

Features

• Run a tap or target
• Pass run options via CLI paramters or JSON/YAML config file.
• Use local file system or S3 for piping the singer stream, storing state, and storing metrics.
• Metric storage, piping, and state storage can be extended / customized by inheriting from the base classes.

Usage

Requires python 3, tested with python 3.7

Install

pip install singer-runner

Run

$ singer-runner 
Usage: singer-runner [OPTIONS] COMMAND [ARGS]...

Options:
  --help  Show this message and exit.

Commands:
  run-tap
  run-target

Concepts

• Pipes
  • Pipes move a stream of Singer messages from tap to target. A pipe could be as simple as a local file, a file in S3, or Kafka.
• State Storage
  • Loads/saves the state JSON.
• Metrics Storage
  • Accepts a stream of Singer Metrics

Programmatic Usage

Singer runner can be used within any python application. The primary functions are in singer_runner.runner including:

• run_tap runs a tap
• run_target runs a target

Classes in the singer_runner.metrics, singer_runner.pipes, and singer_runner.state can be used as arguemnts, along with catalog/config.

singer-runner's Issues

Additional modules required

I followed the instructions in the readme to install from this repo but had to install these modules as well:

• click
• dataclasses

After doing so I was able to run singer-runner successfully.

CLI: Exit with error code on failure in tap or target

The current behavior is that the exit code is always 0. Scripts that depend on this code to determine whether to exit or continue fail to exit as a result.

This is appears to be specific to the CLI. The main functions do return success / failure. The desired behavior is either to exit 1 explicitly or raise an exceptions, which should cause an exit 1.

Webhook Server

In order to support singer-io/getting-started#50 a webserver must accept HTTP requests and pass them on as WEBHOOK messages.

Proposal for how that could work:

• A singer-runner webhook-server command would start an HTTP server.
• POST /webhooks/:pipe-name/:token
  • pipe-name maps to a specific pipe instance ex mailchimp. When a request is successful, a WEBHOOK message is placed on that pipe.
  • token maps to a single token passed in the config or environment. Perhaps a function or class could be passed in to allow for more complex auth.
• run_tap would need to be changed to allow for input pipes.

Unwanted datatype conversion

I am using the singer-runner to run a tap and target combination programmatically. However, by executing it that way, a type conversion happens somewhere along the way.

tap_command = './tap-quickbooks-reports/tap_quickbooks_reports/__init__.py'
p = StdInOutPipe()

run_tap(
            logger = LOGGER,
            tap_command = tap_command,
            tap_config = tap_config,
            pipe = p
)

I've tried to use the StdInOutPipe as well as the FilePipe.

When running my tap in the CLI it works as desired and writes messages in the desired type. When running it through the code above, a float value will be written (with both pipes) as a String value. When modifying my singer schema it works, but I need a float value in the destination. I was not able yet to figure out where/why the conversion happens. I tested my code again, and the problem seems to be introduced in the singer-runner code (I might use it wrong though).

The output of the resulting field looks like this:
"subt_nat_amount": "38.72"

Any idea why this happens?

Support lock files

Support S3 / blobstore lock files with expirations

File path timestamps / templating

For parts of the config / filepaths, allow timestamps and run IDs to be passed through.

Eg

   - s3://my-data-bucket/singer/messages-{timestamp}.txt
   - s3://my-data-bucket/singer/messages-{run_id}.txt

Run ID should be a CLI option that defaults to a UUID if not provided

Handle Singer config file modifications

Some taps modify their own config, the most common reason is to update the refresh tokens for APIs that forceable rotate them.

Support S3 / blobstore runner config locations

Create Pypi project

Create Pypi project and upload to pypi so users can pip install

Config file docs

Support running tap with target via CLI

Running both a tap and target with a memory pipe via CLI