Singer Taps & Extractors
Relevant source files
The following files were used as context for generating this wiki page:
Singer Taps & Extractors form the data ingestion layer of the FLEXT Meltano platform. These components are responsible for extracting data from various sources (databases, APIs, files) and emitting it in the standardized Singer format (JSONL). The system utilizes a mixin-based architecture to provide tap instance creation, stream discovery, and catalog management while maintaining strict protocol compliance for Go ↔ Python bridge operations.
The integration relies on pydantic models for configuration validation and a railway-oriented programming (ROP) approach using Result[T] types to handle operations like stream discovery and synchronization. Sources: src/flext_meltano/services/singer_tap.py:11-20, src/flext_meltano/_protocols/singer.py:9-15, src/flext_meltano/services/abstractions.py:14-25
Architecture and Core Mixins
The architecture leverages Python Mixins to inject Singer-specific capabilities into the FlextMeltano facade. The primary logic is divided between source instance creation, catalog management, and high-level abstractions for Meltano runtime operations.
Component Relationship
The following diagram illustrates how the different mixins and abstractions interact to provide a unified tap interface.
The diagram shows the hierarchy where ServiceBase provides logging and basic infrastructure, while specialized mixins handle tap creation and catalog lifecycle.
Sources: src/flext_meltano/services/singer_tap.py:17-20, src/flext_meltano/services/singer_catalog.py:19-25, src/flext_meltano/services/abstractions.py:14-15
Configuration Models
Taps are configured using structured Pydantic models that ensure type safety and sensitive data protection. The TapConfig model handles specific extraction settings, while DataSourceConfig provides a generic wrapper used during instance creation.
Tap Configuration Fields
| Field | Type | Description |
|---|---|---|
tap_type | str | The identifier for the tap (e.g., tap-postgres). |
connection_config | t.JsonMapping | Raw connection parameters (protected during serialization). |
stream_config | t.JsonMapping | Specific configurations for individual streams. |
tap_version | str | Version of the tap plugin (defaults to "latest"). |
Sources: src/flext_meltano/_models/sources.py:20-33
Sensitive Data Protection
All configuration models implement a field serializer for connection_config that utilizes FlextMeltanoModelsCore.protect_sensitive_config to mask credentials before logging or serialization. Sources: src/flext_meltano/_models/sources.py:53-56
Stream Discovery and Synchronization
Discovery is performed via the Meltano CLI or through direct Singer SDK calls, depending on the context. The FlextMeltanoAbstractions class manages the interaction with the Meltano binary to list and select streams.
Discovery Flow
The process involves invoking the meltano select --list --all command and parsing the resulting output into a registry of StreamDefinition models.
This flow describes how abstractions interact with the Meltano CLI to populate the internal stream registry.
Sources: src/flext_meltano/services/abstractions.py:35-69
Synchronization Logic
Synchronization uses the meltano elt command. It allows for single-stream sync by applying the --select option to the execution arguments.
def sync_stream(
self,
tap_instance: m.Meltano.TapInstance,
stream_name: str,
target_config: m.Meltano.TargetConfig | None = None,
) -> p.Result[t.JsonMapping]:
# ... logic to determine loader name ...
cmd_args = [
c.Meltano.CMD_ELT,
tap_instance.tap_type,
loader_name,
c.Meltano.CMD_SELECT_OPTION,
stream_name,
]
cmd_result = self._run_meltano(cmd_args)
# ... handle mapping to COMPLETED/FAILED status ...
Sources: src/flext_meltano/services/abstractions.py:82-113
Singer Catalog Management
The FlextMeltanoSingerCatalogMixin manages the lifecycle of Singer catalogs, which define the schemas and replication methods for data extraction.
Key Catalog Operations
- Discovery: Discovers available streams from a running
SingerTapinstance and updates the cached_singer_catalog. Sources: src/flext_meltano/services/singer_catalog.py:32-46 - Persistence: Catalogs can be saved to or loaded from disk using atomic write operations. Sources: src/flext_meltano/services/singer_catalog.py:64-100
- Selection: Allows filtering the catalog to include only specific streams for a sync operation. Sources: src/flext_meltano/services/singer_catalog.py:102-121
Singer Protocol Utilities
The FlextMeltanoUtilitiesSinger provides static methods for emitting protocol-compliant messages to stdout. This ensures that any tap implemented within the FLEXT framework adheres to the expected Singer message formats.
Message Emission Methods
| Method | Purpose | Key Parameters |
|---|---|---|
emit_schema | Emits a SCHEMA message. | stream_name, schema, key_properties. |
emit_record | Emits a RECORD message. | stream_name, record, time_extracted. |
emit_state | Emits a STATE message. | value (bookmark payload). |
Sources: src/flext_meltano/_utilities/singer.py:23-95
Conclusion
Singer Taps & Extractors provide a robust, type-safe mechanism for data extraction by wrapping standard Singer and Meltano operations in high-level Python abstractions. Through the use of Pydantic models and specialized mixins, the system ensures configuration consistency, secure credential handling, and reliable stream orchestration.