Meltano Project SDK & Bridge
Relevant source files
The following files were used as context for generating this wiki page:
Introduction
The Meltano Project SDK & Bridge provides a production-ready integration layer between the FLEXT ecosystem and the Meltano data integration engine. It functions as a Go ↔ Python bridge, allowing Go-based services to orchestrate complex data pipelines using Meltano, Singer, and DBT technologies. This module encapsulates the complexity of subprocess orchestration, environment management, and project lifecycle operations into a unified Python API.
The primary purpose of the bridge is to translate high-level service requests into concrete Meltano CLI commands while maintaining enterprise-grade standards for error handling, structured logging, and type safety. It ensures that data operations, such as extraction (Taps) and loading (Targets), are performed within a validated project structure.
Sources: src/flext_meltano/README.md:16-18, src/flext_meltano/README.md:92-97, src/flext_meltano/services/meltano_project_sdk.py:1-7
Architecture and Core Components
The architecture is organized into layers that separate project management, command execution, and runtime utilities. The system relies on a Foundation Layer for configuration and a Bridge Integration Layer for orchestration.
Meltano Project SDK Management
The FlextMeltanoProjectManager mixin provides direct integration with the meltano.core.project.Project SDK. It manages the lifecycle of a Meltano project, including initialization, loading existing projects, and extracting plugin metadata.
The manager tracks the project root and current SDK project state using private attributes _sdk_project_root and _sdk_project.
Sources: src/flext_meltano/services/meltano_project_sdk.py:16-25, src/flext_meltano/services/meltano_project_sdk.py:84-118
Integration Bridge & Abstractions
The FlextMeltanoAbstractionsBase class serves as the functional bridge, wrapping in-process Meltano operations with specialized Result[T] containers for railway-oriented programming. It handles the mapping of abstract pipeline requests to executable commands like meltano elt and meltano add.
Sources: src/flext_meltano/services/abstractions_base.py:21-25, src/flext_meltano/services/abstractions_base.py:80-90
Project Lifecycle Operations
The SDK provides explicit methods for managing the physical and logical state of Meltano projects.
Project Initialization and Loading
The system supports two primary entry points for project interaction:
- Initialization: Creates a new directory structure and instantiates a Meltano
Projectobject. - Loading: Attaches to an existing project on the file system, verifying the directory presence before instantiation.
| Method | Purpose | Result State |
|---|---|---|
initialize_sdk_project | Prepares directory and creates new project | c.Meltano.ProjectSdkState.INITIALIZED |
load_sdk_project | Loads existing project from Path | c.Meltano.ProjectSdkState.LOADED |
Sources: src/flext_meltano/services/meltano_project_sdk.py:46-64, src/flext_meltano/services/meltano_project_sdk.py:66-82
Runtime Command Normalization
The FlextMeltanoUtilitiesRuntime class provides static helpers to ensure CLI arguments are correctly formatted for the Meltano engine. This includes stripping whitespace, filtering empty parts, and resolving command aliases.
Data Flow and Protocols
The system defines strict protocols for projects and adapters to ensure interoperability between different pipeline components.
Project and Adapter Protocols
The FlextMeltanoProtocolsProject defines the Project, Adapter, and IndexedPlugin protocols. The Adapter protocol is critical for data source/sink interactions, requiring implementations for connect, disconnect, and connection status checks.
Sources: src/flext_meltano/_protocols/project.py:17-48
Plugin Discovery Flow
Plugin discovery involves querying the project or external hubs to identify available Extractors and Loaders. The bridge extracts plugin names, namespaces, and variants to build a canonical discovery payload.
Configuration and Settings
Orchestration is governed by the FlextMeltanoSettings model, which uses environment variables with the FLEXT_MELTANO_ prefix.
| Setting | Env Alias | Default | Description |
|---|---|---|---|
project_root | MELTANO_PROJECT_ROOT | Path() | Root directory of the Meltano project |
environment | MELTANO_ENVIRONMENT | development | Active runtime environment (dev, testing, prod) |
config_dir | N/A | .meltano | Configuration storage directory |
meltano_version | N/A | 3.9.1 | Required Meltano binary version |
Sources: src/flext_meltano/settings.py:16-60, src/flext_meltano/_constants/base.py:30-40
Implementation Example: Subprocess Execution
The following logic demonstrates how the bridge executes a command and processes the result into a standardized payload.
def _run_meltano(self, args: t.StrSequence) -> p.Result[str]:
cwd = u.Meltano.resolve_project_root(self.settings)
run_result = FlextMeltanoExecutorBase().execute_meltano_command(
list(args), _cwd=cwd
)
if run_result.failure:
return r[str].fail_op("Meltano command", run_result.error)
completed = run_result.value
if completed.exit_code != 0:
return r[str].fail(completed.error.strip() or "meltano failed")
return r[str].ok(completed.output.strip())
Sources: src/flext_meltano/services/abstractions_base.py:27-44
Summary
The Meltano Project SDK & Bridge serves as the foundational orchestration engine for FLEXT data integration. By combining the meltano.core SDK with robust subprocess management and utility-based normalization, it provides a stable interface for Go services to trigger complex Python-based data pipelines. The use of strict protocols and settings-driven configuration ensures that integration remains consistent across development, testing, and production environments.