Custom Sink Implementation
Relevant source files
The following files were used as context for generating this wiki page:
The Custom Sink Implementation in flext-target-oracle-wms facilitates the transfer of Singer protocol data into Oracle WMS Cloud SaaS. It serves as a specialized adapter that bridges the Singer SDK's sink logic with Oracle-specific WMS patterns, ensuring that data is normalized, transformed, and loaded according to WMS requirements.
Sources: src/flext_target_oracle_wms/_utilities/service_runtime.py:15-18, src/flext_target_oracle_wms/api.py:27-33
Architecture and Components
The sink implementation utilizes a multi-layered approach involving a service facade, a runtime adapter, and internal helpers to manage the lifecycle of a data stream.
FlextTargetOracleWmsService
The top-level orchestrator responsible for initializing the sink through the create_sink method. It acts as a facade that utilizes settings overrides to configure the underlying runtime.
Sources: src/flext_target_oracle_wms/api.py:20-36
FlextTargetOracleWmsServiceRuntime.Sink
This internal class acts as the concrete implementation of m.Meltano.SingerSinkBase. It holds a reference to a _runtime_target and is responsible for processing individual records and batches.
Sources: src/flext_target_oracle_wms/_utilities/service_runtime.py:17-25
WMS Integration Protocols
The system defines specific protocols to ensure consistency in data loading and transformation:
- WmsDataLoading: Defines the contract for loading a sequence of JSON records into a sink.
- DataTransformation: Defines the contract for converting a standard JSON record into a WMS-compliant shape.
Sources: src/flext_target_oracle_wms/protocols.py:17-38
Component Relationship Diagram
The following diagram illustrates the relationship between the service facade, the sink runtime, and the underlying WMS target.
Data Flow and Processing Logic
Data processing follows a strict sequence of normalization, validation, and transformation before the final load into Oracle WMS.
Record Processing Flow
When process_record is invoked, the sink performs the following steps:
- Normalization: Converts the incoming record into a standard JSON mapping.
- Validation: Validates the record against the
SingerRecordMessagemodel. - Transformation: The runtime target handles the record, applying WMS-specific transformations (e.g., uppercasing keys).
- Error Handling: If the runtime rejects the record, a
RuntimeErroris raised with the specific error message.
Schema Normalization
Before a sink can be created, the schema must be normalized to meet the WMS runtime contract. This involves converting complex types (like Path objects) into strings.
Sources: src/flext_target_oracle_wms/_utilities/service_runtime.py:100-105
Key Implementation Details
Data Transformation Patterns
The WMSDataTransformer handles the mapping of Singer record payloads. A critical part of this transformation is ensure keys are uppercased to match Oracle WMS requirements.
Sources: src/flext_target_oracle_wms/_utilities/helpers.py:82-110, tests/unit/test_sinks.py:92-98
Configuration Validation
The sink initialization requires a specific set of authentication fields. The Validation helper ensures these are present before processing begins.
| Parameter | Required | Description |
|---|---|---|
base_url | Yes | The URL of the Oracle WMS Cloud SaaS instance. |
username | Yes | WMS authentication username. |
password | Yes | WMS authentication password. |
load_method | No | Method of loading (e.g., APPEND_ONLY, UPSERT, MERGE). |
Sources: src/flext_target_oracle_wms/_utilities/helpers.py:19-35, examples/01_basic_usage.py:28-32
Sink Initialization Example
The FlextTargetOracleWmsServiceRuntime provides the factory logic for creating the sink with a fully initialized runtime target.
@classmethod
def create_sink(
cls,
*,
stream_name: str,
schema: t.JsonMapping,
target_config: t.JsonMapping,
) -> p.Meltano.SingerDrainSink:
"""Create the service-level Singer sink adapter."""
normalized_target_config = u.normalize_to_json_mapping(target_config)
runtime_target = u.TargetOracleWms.Target(normalized_target_config)
normalized_schema = cls.normalize_flat_schema(schema)
# ... handle schema message and return Sink instance ...
return cls.Sink.create(...)
Sources: src/flext_target_oracle_wms/_utilities/service_runtime.py:72-98
Summary
The Custom Sink Implementation utilizes the FlextTargetOracleWmsServiceRuntime to bridge Singer protocol messages with Oracle WMS operations. By delegating processing to a Target runtime and utilizing standardized WMSDataTransformer and WMSTableManager components, the system ensures that data is correctly formatted for enterprise-grade SaaS integration. The use of strict protocols like WmsDataLoading ensures that the sink remains compliant with the project's architectural standards.
Sources: src/flext_target_oracle_wms/api.py:38, src/flext_target_oracle_wms/protocols.py:21-35