Singer Message Processors
Relevant source files
The following files were used as context for generating this wiki page:
Singer Message Processors within the flext-target-oracle-oic project are responsible for handling and transforming incoming Singer-formatted record messages into structures compatible with the Oracle Integration Cloud (OIC) API. These processors reside within the singer sub-package and serve as the primitive building blocks for stream-specific record manipulation.
The architecture leverages a lazy-loading mechanism for its exports, ensuring that processor components are only loaded when explicitly required by the target's sink operations. While the processors.py file defines the primitives, the actual execution and orchestration are managed through the project's Sink classes and service runtime adapters.
Sources: src/flext_target_oracle_oic/singer/processors.py, src/flext_target_oracle_oic/singer/__init__.py, src/flext_target_oracle_oic/target.py
Architecture and Data Flow
The data flow begins when the FlextTargetOracleOic entry point receives Singer messages. The target resolves the appropriate sink class based on the stream name. The sink then utilizes message processors to validate and transform the RECORD messages before they are passed to the OIC API service facade.
Message Processing Pipeline
The following diagram illustrates the flow from a Singer message to the OIC-specific sink processing.
The pipeline ensures that generic JSON payloads are mapped to validated Pydantic models for Oracle OIC entities.
Sources: src/flext_target_oracle_oic/target.py:59-71, src/flext_target_oracle_oic/utilities.py:38-66
Core Components
Sink Implementation
The sinks act as the primary consumers of processed messages. FlextTargetOracleOicBaseSink provides the foundational logic for batch and record processing.
| Sink Class | Target Stream | Description |
|---|---|---|
FlextTargetOracleOicConnectionsSink | connections | Processes OIC connection definitions. |
FlextTargetOracleOicIntegrationsSink | integrations | Handles OIC integration flows and metadata. |
FlextTargetOracleOicPackagesSink | packages | Manages OIC package imports and exports. |
FlextTargetOracleOicLookupsSink | lookups | Processes lookup table data. |
Sources: src/flext_target_oracle_oic/target.py:22-54, tests/unit/test_e2e_sinks.py:44-53
Service Runtime Adapters
The FlextTargetOracleOicServiceRuntime acts as an intermediary, creating service-level Singer sink adapters. It normalizes target configurations and initializes the appropriate sink class with the provided schema.
This runtime structure allows the facade to remain thin while offloading the complexity of sink initialization and configuration normalization.
Sources: src/flext_target_oracle_oic/_utilities/service_runtime.py:17-43, src/flext_target_oracle_oic/api.py:30-41
Logic and Validation
Message processing relies on specific factory helpers to transform raw JSON data into OIC-compliant models. This is handled by the FlextTargetOracleOicUtilities class, which contains logic for:
- Model Validation: Using Pydantic to validate that incoming records match the required structure for
OICConnection,OICIntegration,OICPackage, orOICLookup. - Configuration Validation: Ensuring the target environment contains the necessary OAuth2 credentials (
base_url,oauth_client_id,oauth_client_secret). - Authentication Handling: An internal
Authenticatorclass manages the generation and refreshing of OAuth2 Bearer tokens used during the transmission of processed messages.
Sources: src/flext_target_oracle_oic/utilities.py:25-36, src/flext_target_oracle_oic/utilities.py:38-66, src/flext_target_oracle_oic/utilities.py:68-124
Example: Record Processing Logic
The base sink provides a standard process_record hook that logs metadata before deeper processing occurs.
# src/flext_target_oracle_oic/target.py:31-38
@override
def process_record(
self,
record: t.MutableJsonMapping,
context: t.MutableJsonMapping,
) -> None:
"""Default sink behavior: log incoming record metadata."""
_ = context
self.logger.debug("Processing OIC record", keys=str(list(record.keys())))
Summary
Singer Message Processors in this repository provide a structured way to ingest data from Singer sources and prepare it for Oracle Integration Cloud. By utilizing stream-specific sinks and a central service runtime, the system ensures that every record is validated against strict OIC models and delivered with proper OAuth2 authentication. The use of lazy exports and specialized utilities maintains a modular architecture capable of handling various OIC entity types like connections, integrations, and packages.