Catalog & Schema Translation
Relevant source files
The following files were used as context for generating this wiki page:
The Catalog & Schema Translation system in FLEXT Meltano is responsible for managing the structure of data streams during extraction and loading processes. It serves as a bridge between the Singer protocol's metadata requirements and the Meltano orchestration layer, ensuring that data schemas are accurately discovered, cached, and manipulated to facilitate precise data integration workflows.
This module provides the necessary abstractions for discovering streams from Singer taps, loading and saving catalog definitions to the filesystem, and selecting specific subsets of data for synchronization. It leverages a Mixin-based architecture to provide these capabilities to the main FlextMeltano facade, enabling high-level operations on underlying Singer catalogs.
Sources: src/flext_meltano/services/singer_catalog.py:1-18, src/flext_meltano/services/__init__.py:78-80
Catalog Management Architecture
Catalog management is implemented primarily through the FlextMeltanoSingerCatalogMixin. This class provides stateful management of a SingerCatalog object, which contains the definitions for various data streams, including their schemas and metadata.
Data Flow for Catalog Discovery
The discovery process involves interacting with a Singer tap instance to extract the available stream definitions. The resulting catalog is then cached internally within the service mixin.
The diagram shows the logical flow when discover_catalog_streams is invoked, detailing the transition from tap discovery to internal state updates.
Sources: src/flext_meltano/services/singer_catalog.py:27-43
Key Service Methods
| Method | Description | Return Type |
|---|---|---|
discover_catalog_streams | Triggers a discovery operation on a provided Singer tap to map available streams. | p.Result[m.Meltano.SingerCatalog] |
fetch_stream_schema | Retrieves the JSON schema definition for a specific stream from the current catalog. | p.Result[t.JsonMapping] |
load_catalog | Deserializes a catalog from a JSON file located on the filesystem. | p.Result[m.Meltano.SingerCatalog] |
save_catalog | Performs an atomic write of the current catalog state to a specified file path. | p.Result[None] |
select_streams | Filters the cached catalog to include only a specified list of stream names. | p.Result[m.Meltano.SingerCatalog] |
Sources: src/flext_meltano/services/singer_catalog.py:27-111
Schema Translation and Validation
FLEXT Meltano utilizes Pydantic-based models to ensure that catalog payloads conform to the expected Singer structure. This validation layer prevents malformed metadata from entering the pipeline.
Catalog Model Structure
The system defines a formal structure for catalogs and their constituent streams. A SingerCatalog is essentially a collection of stream entries, each containing the stream name, a unique ID, and the schema definition.
This diagram represents the relationship between the catalog container and individual stream definitions as validated by the internal models.
Sources: tests/unit/test_plugin_protocols.py:44-51, src/flext_meltano/_models/singer_catalog.py
Schema Retrieval Logic
When fetch_stream_schema is called, the system iterates through the internal _singer_catalog.streams list. It matches the requested stream_name against the entry.stream property and returns the associated schema_definition.
Sources: src/flext_meltano/services/singer_catalog.py:45-56
Persistence and CLI Translation
The FlextMeltanoSingerCliTranslator handles the conversion of internal catalog structures into formats suitable for CLI interactions. Meanwhile, the persistence layer ensures that catalogs can be managed as files for incremental synchronization and debugging.
File Operations
- Loading: Uses
u.Cli.files_read_json_modelto safely load JSON into the Pydantic model. - Saving: Uses
cli.atomic_write_text_fileto ensure data integrity during writes, preventing partial file corruption.
Sources: src/flext_meltano/services/singer_catalog.py:58-91
Tap and Target Configuration Models
Configurations for Taps and Targets include metadata that identifies the component type and version, alongside the connection and stream configurations.
| Field | Type | Description |
|---|---|---|
tap_type / target_type | str | The identifier for the Singer component (e.g., "tap-postgres"). |
connection_config | JsonMapping | Sensitive connection parameters (host, user, etc.). |
stream_config | JsonMapping | Parameters specific to individual streams. |
batch_size | int | (Target only) The number of records to process in one batch. |
Sources: src/flext_meltano/_models/sources.py:17-104
Summary
The Catalog & Schema Translation module provides a robust foundation for managing Singer metadata within the FLEXT ecosystem. By combining strict Pydantic model validation (SingerCatalog) with flexible service mixins (FlextMeltanoSingerCatalogMixin), it enables developers to discover, filter, and persist data schemas with high confidence. This system ensures that the orchestrator (Meltano) and the data movers (Singer Taps/Targets) remain synchronized regarding the structure of the data being integrated.
Sources: src/flext_meltano/services/singer_catalog.py:1-18, src/flext_meltano/_models/sources.py:11-15