Custom Tap & Target Services
Relevant source files
The following files were used as context for generating this wiki page:
Custom Tap and Target services within the FLEXT Meltano platform provide an enterprise-grade abstraction layer over the standard Singer SDK. These services facilitate the creation of specialized data extractors (Taps) and loaders (Targets) by providing standardized base classes that handle lifecycle management, CLI dispatching, and connection protocols. The system ensures technical consistency through a Railway-Oriented Programming (ROP) pattern where operations return Result types containing either success values or structured failure data.
The architecture is designed for extensibility, allowing developers to focus on domain-specific extraction or loading logic while inheriting robust infrastructure for stream discovery, synchronization, and batch processing. This module serves as the primary bridge between raw Singer protocol implementations and the FLEXT ecosystem's orchestration capabilities.
Sources: src/flext_meltano/services/consumer_bases/tap_service_base.py, src/flext_meltano/services/consumer_bases/target_service_base.py, src/flext_meltano/api.py
Service Architecture and Hierarchy
The system uses a Mixin-based approach and inheritance to build complex service facades. The FlextMeltano class acts as a global facade, aggregating capabilities from various service abstractions.
Core Class Structure
The following diagram illustrates the relationship between the base service classes and the specialized consumer bases used for custom implementations.
The diagram shows the specialization of services into Tap and Target hierarchies.
Sources: src/flext_meltano/services/consumer_bases/tap_service_base.py:22-38, src/flext_meltano/services/consumer_bases/target_service_base.py:25-45
Custom Tap Services
Custom Taps are responsible for extracting data from source systems. The FlextMeltanoTapServiceBase provides the lifecycle management for these components, wrapping a singer_sdk.Tap instance.
Tap Lifecycle and Operations
A custom Tap service manages the following primary operations:
- Discovery: Identifying available data streams and their schemas.
- Sync: Iterating through streams and emitting records in Singer format.
- Connection: Managing the session with the source data store.
The flow represents how the Tap service handles CLI commands to trigger discovery or synchronization.
Sources: src/flext_meltano/services/consumer_bases/tap_service_base.py:65-104
Key Functions
| Function | Purpose | Return Type |
|---|---|---|
create_tap_instance | Abstract factory method for the domain-specific Tap class. | p.Meltano.SingerTapInstance |
run_discover | Executes stream discovery and returns a list of stream names. | p.Result[t.StrSequence] |
run_sync | Executes the full synchronization process for all selected streams. | p.Result[str] |
cli_main | Entry point for command-line execution, dispatching to Singer bridge. | int |
Sources: src/flext_meltano/services/consumer_bases/tap_service_base.py:53-104
Custom Target Services
Custom Targets act as sinks for data emitted by Taps. FlextMeltanoTargetServiceBase handles the complexity of sink management and batch record processing.
Sink and Record Management
Targets use "Sinks" to handle the actual writing of data to a destination. The base service manages a registry of sinks, ensuring that a sink is created for every unique stream encountered during a sync operation.
The sequence shows how records are routed to specific sinks and eventually flushed to the destination.
Sources: src/flext_meltano/services/consumer_bases/target_service_base.py:75-125
Data Processing Methods
process_record: Validates the incoming record against the expected schema and passes it to the appropriate sink.process_batch: Iterates over a sequence of records, processing each through the ROP pipeline.flush: Triggers thestart_drainandprocess_batchcycle on sinks to commit data to the target store.
Sources: src/flext_meltano/services/consumer_bases/target_service_base.py:104-143
Configuration Models
Configurations for Taps and Targets are strictly typed using Pydantic-based models to ensure data integrity and security (e.g., protecting sensitive credentials).
| Model | Key Fields | Validation Logic |
|---|---|---|
TapConfig | tap_type, connection_config, stream_config | Ensures tap_type and connection_config are non-empty. |
TargetConfig | target_type, connection_config, batch_size | Validates target_type is present. |
DataSourceConfig | source_type, connection_config, source_version | General purpose model for source connectivity. |
Sources: src/flext_meltano/_models/sources.py:20-136
Sensitive Data Protection
The models utilize a serialize_connection_config method via FlextMeltanoModelsCore.protect_sensitive_config to ensure that connection details (like passwords or tokens) are handled securely during serialization.
Sources: src/flext_meltano/_models/sources.py:53, 99, 140
Implementation Example: Service Creation
The FlextMeltano facade provides high-level methods to instantiate these specialized services.
# API example for creating services
from flext_meltano import meltano
# Create a tap instance
tap_result = meltano.tap("tap-postgres", host="localhost", port=5432)
# Create a target instance
target_result = meltano.target("target-snowflake", account="xyz123")
Sources: src/flext_meltano/api.py:44-52
Conclusion
Custom Tap & Target Services in FLEXT Meltano provide a standardized, type-safe framework for data integration. By abstracting the Singer SDK's boilerplate code into FlextMeltanoTapServiceBase and FlextMeltanoTargetServiceBase, the platform allows for rapid development of robust connectors that integrate seamlessly with the broader Meltano ecosystem. The use of the Result pattern ensures that these services remain resilient and provide clear diagnostic information during pipeline failures.