Singer Python SDK

Warnings

• breaking Any non-Singer output (e.g., plain text via `print()`) to `stdout` will break downstream Singer targets. Singer targets expect `stdout` to contain only line-delimited JSON messages conforming to the Singer protocol.
  Fix: Use `singer.get_logger()` for all informational and debug output. By default, `singer.get_logger()` writes to `stderr`, ensuring `stdout` remains clean for Singer messages. Avoid all other `print()` statements.
• gotcha Incorrect or untimely state management can lead to unreliable incremental syncs, data loss, or reprocessing. State messages should accurately reflect processed data and be emitted only *after* all corresponding records for a given checkpoint have been successfully written.
  Fix: Thoroughly understand the Singer spec for state management. Ensure state is written atomically and reliably, typically after a batch of records has been confirmed processed, to correctly bookmark progress.
• gotcha `singer-python` provides utilities for writing schemas but does not automatically handle complex schema evolution logic (e.g., column renames, type changes requiring migration). Taps must adhere to the Singer spec for schema changes.
  Fix: Consult the Singer spec regarding schema evolution. Only introduce additive changes (new fields) to a stream's schema or utilize the `activate_version` message for breaking changes, ensuring the target supports it. Plan schema changes carefully.
• gotcha For new projects requiring a more opinionated framework with robust CLI parsing, test helpers, and higher-level abstractions for Singer, consider using `meltano-sdk` which builds upon the Singer protocol, rather than `singer-python` directly.
  Fix: Evaluate your project's needs: `singer-python` for barebones Singer protocol adherence and maximum control; `meltano-sdk` for a more complete, feature-rich development experience for taps and targets.

Install

• pip install singer-python Install latest version

Imports

• singer

  import singer

• get_logger

  singer.get_logger()

• write_schema

  singer.write_schema(...)

• write_record

  singer.write_record(...)

• write_state

  singer.write_state(...)

• cli

  import singer.cli

  Used for command-line interface utilities like `singer.cli.main` or `singer.cli.with_cli` decorator for taps and targets.

Quickstart

This quickstart demonstrates the core functionality of `singer-python`: emitting schema, record, and state messages to standard output, which is the standard mechanism for Singer taps.

import singer
import json
import sys

# Get a Singer logger (logs to stderr by default)
LOGGER = singer.get_logger()

# 1. Define a schema for a stream
stream_name = "users"
schema = {
    "type": "object",
    "properties": {
        "id": {"type": "integer", "key_properties": ["id"]},
        "name": {"type": "string"},
        "email": {"type": "string", "format": "email"}
    }
}
key_properties = ["id"]

# 2. Write the schema message to stdout
# This is how a Tap declares the structure of data it will send
singer.write_schema(
    stream_name=stream_name,
    schema=schema,
    key_properties=key_properties
)
LOGGER.info(f"Schema written for stream '{stream_name}'")

# 3. Write record messages to stdout
# These are the actual data rows
records = [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob", "email": "bob@example.com"}
]

for record in records:
    singer.write_record(
        stream_name=stream_name,
        record=record
    )
    LOGGER.debug(f"Record written for '{stream_name}': {record['id']}")

# 4. Write a state message to stdout
# This allows a Tap to checkpoint its progress for incremental syncs
state = {"bookmarks": {stream_name: {"last_id": records[-1]["id"]}}}
singer.write_state(state)
LOGGER.info(f"State written: {state}")

# The actual Singer messages are written to stdout.
# To see them, run this script and redirect stdout:
# python your_script.py > output.jsonl