Target Hotglue

Common errors

• ModuleNotFoundError: No module named 'target_hotglue'

  cause The `target-hotglue` library has not been installed or is not accessible in the current Python environment.
  fix Install the library using pip: `pip install target-hotglue`

• ERROR: Package 'target-hotglue' requires Python '>=3.7.1, <3.11' but the running Python is 3.11.X

  cause The Python version being used is not compatible with the `target-hotglue` library's requirements.
  fix Switch to a supported Python version (e.g., 3.7-3.10). You can use tools like `pyenv` or `conda` for environment management. Example: `pyenv local 3.10.12` or create a new virtual environment with `python3.10 -m venv .venv`.

• KeyError: 'some_required_setting'

  cause Your target's configuration (e.g., in `config.json` or environment variables) is missing a required field defined in your `config_jsonschema` or expected by the target's logic.
  fix Review your target's `config_jsonschema` and ensure all required configuration keys are present and correctly formatted in your configuration file or environment variables. Example: `echo '{"output_file": "my_output.jsonl"}' > config.json`

Warnings

• gotcha `target-hotglue` currently has a Python version constraint of `<3.11`. Attempting to install or run it with Python 3.11 or newer will result in installation errors.
  Fix: Ensure your Python environment is between 3.7.1 and 3.10.x. Consider using `pyenv` or `conda` to manage Python versions. Example: `pyenv install 3.10.12 && pyenv local 3.10.12`
• gotcha `target-hotglue` is built on top of `singer-sdk`. Breaking changes or significant updates in `singer-sdk` (especially between major or even minor versions like `0.x` to `0.y`) can indirectly impact your `target-hotglue` implementation, even if `target-hotglue` itself hasn't released a new version.
  Fix: Always check the release notes for `singer-sdk` when upgrading either library. Pin specific versions of `singer-sdk` in your project's dependencies to prevent unexpected behavior: `pip install target-hotglue singer-sdk==0.X.Y`
• gotcha Proper Singer state management, inherited from `singer-sdk`, is crucial for correct data processing (avoiding reprocessing or data loss). Misunderstanding how state messages are handled or misconfiguring the target's state backend can lead to integrity issues.
  Fix: Thoroughly review `singer-sdk`'s documentation on state management. Ensure your target correctly reads and writes state messages, especially when using custom state backends or when running in environments like Hotglue that manage state externally.

Install

• pip install target-hotglue Install latest version

Imports

• Target

  from target_hotglue.target import Target

  Main class to inherit from when creating a custom target.
• Stream

  from target_hotglue.streams import Stream

  Base class for defining how records for a specific stream are processed.

Quickstart

This quickstart demonstrates how to define a custom target and stream using `target-hotglue` and simulate processing Singer messages. In a production environment, `target-hotglue` typically reads Singer messages from `stdin` via its `main` entry point. This example writes processed records to a local JSONL file defined by the `output_file` configuration setting, which can be overridden by the `TARGET_HOTGLUE_OUTPUT_FILE` environment variable.

import os
import json
from pathlib import Path
from typing import Dict, Any, List

from target_hotglue.target import Target
from target_hotglue.streams import Stream

# --- 1. Define your custom target and streams ---

# Define a custom stream where records will be written
class MyCustomStream(Stream):
    name = "my_custom_stream"
    schema = {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "string"}}}
    key_properties = ["id"]

    def write_record(self, record: Dict[str, Any]) -> None:
        """
        Processes a single record.
        In a real target, this is where you'd write to your destination system.
        For this example, we'll write to a local JSONL file.
        """
        output_file_path = Path(self.target_config.get("output_file", "quickstart_output.jsonl"))
        with output_file_path.open("a") as f:
            f.write(json.dumps(record) + "\n")
        self.logger.info(f"Wrote record to {output_file_path}: {record}")

# Define your custom target, linking it to your stream(s)
class MyCustomTarget(Target):
    name = "target-quickstart"
    # Define a simple configuration schema for the target
    config_jsonschema = {
        "type": "object",
        "properties": {
            "output_file": {"type": "string", "default": "quickstart_output.jsonl", "description": "File to write records to"}
        },
        "required": []
    }

    # Register your custom streams with the target
    # The target will route records for 'my_custom_stream' to MyCustomStream
    def get_available_streams(self) -> List[Stream]:
        return [MyCustomStream(target=self)]

# --- 2. How to run your target (simulated stdin) ---

# This simulates the input a Singer tap would send to target_hotglue
sample_singer_messages = [
    {"type": "SCHEMA", "stream": "my_custom_stream", "schema": MyCustomStream.schema, "key_properties": MyCustomStream.key_properties},
    {"type": "RECORD", "stream": "my_custom_stream", "record": {"id": 1, "value": "Hello from Singer!"}},
    {"type": "RECORD", "stream": "my_custom_stream", "record": {"id": 2, "value": "Another record."}},
    {"type": "STATE", "value": {"bookmarks": {"my_custom_stream": {"last_id": 2}}}}
]

def run_quickstart():
    # Clean up previous run's output file
    output_path = Path(os.environ.get("TARGET_HOTGLUE_OUTPUT_FILE", "quickstart_output.jsonl"))
    if output_path.exists():
        output_path.unlink()

    # Initialize the target with a configuration (can be from env vars or a file)
    # Using os.environ.get for config values as requested.
    config = {
        "output_file": os.environ.get("TARGET_HOTGLUE_OUTPUT_FILE", "quickstart_output.jsonl")
    }
    target_instance = MyCustomTarget(config=config)

    # Manually process sample messages (this replaces target_instance.listen_for_messages() for demo)
    print(f"Processing {len(sample_singer_messages)} sample messages...")
    for message in sample_singer_messages:
        target_instance._process_message(message) # Use internal method for demo simplicity

    print(f"\nQuickstart finished.")
    if output_path.exists():
        print(f"Content written to '{output_path}':")
        print(output_path.read_text())
    else:
        print("No output file was created. Check stream configuration.")

if __name__ == "__main__":
    run_quickstart()