Stream Processing
Relevant source files
The following files were used as context for generating this wiki page:
Stream Processing in the flext-target-oracle-wms project handles the ingestion, transformation, and management of Singer-compliant data streams for loading into Oracle WMS Cloud. It utilizes a structured pipeline to parse incoming JSON messages, validate schemas, and transform records into a format compatible with Oracle database requirements, such as uppercasing keys and converting data types.
The system is designed for high-performance batch operations and mission-critical reliability, supporting concurrent stream processing and robust error handling patterns like retries and circuit breakers.
Sources: src/flext_target_oracle_wms/_utilities/client.py:92-99, examples/README.md:104-120
Architecture and Data Flow
The stream processing architecture relies on a central Target runtime that coordinates several specialized components. The data flow follows the Singer protocol, processing SCHEMA, RECORD, and STATE messages in sequence.
Processing Pipeline
- Line Parsing: Raw input lines (typically from
stdin) are parsed and validated as Singer messages. - Message Dispatching: The
Targetidentifies the message type and routes it to the appropriate handler. - Metadata Management:
SCHEMAmessages trigger stream registration in theCatalogManagerandWMSTableManager. - Record Transformation:
RECORDmessages are passed to theStreamProcessor, which coordinates with theWMSDataTransformerto map fields and convert types. - Persistence/Loading: Transformed data is prepared for batch insertion into Oracle WMS tables.
Sources: src/flext_target_oracle_wms/_utilities/client.py:146-192, tests/unit/test_workflow.py:64-79
Component Relationship
The following diagram illustrates the relationship between the core stream processing classes:
Core Components
StreamProcessor
The StreamProcessor acts as the orchestrator for individual stream lifecycles. It is responsible for initializing stream metadata in the table manager and delegating record transformation.
initialize_stream: Registers the stream name and ensures the corresponding Oracle table is mapped (e.g., converting "inventory" to "INVENTORY").process_record: Validates the record against its registered schema and invokes the transformation logic.
Sources: src/flext_target_oracle_wms/_utilities/client.py:61-90, tests/unit/test_stream.py:48-60
WMSDataTransformer & TypeConverter
These components ensure data compatibility with Oracle WMS. A key requirement is that all record keys are transformed to uppercase to match Oracle's standard naming conventions.
| Component | Description |
|---|---|
WMSDataTransformer | Maps Singer record fields to Oracle table columns, ensuring key casing and delegating value conversion. |
WMSTypeConverter | Converts Singer types (string, integer, object, etc.) to Oracle-friendly formats. Complex types like object or array are serialized to JSON strings. |
Sources: src/flext_target_oracle_wms/_utilities/helpers.py:38-51, src/flext_target_oracle_wms/_utilities/helpers.py:67-105
Catalog and Table Management
The CatalogManager stores in-memory metadata for streams, while the WMSTableManager maintains the mapping between Singer stream IDs and Oracle physical table names.
Message Handling Logic
The processing loop handles different Singer message types with specific behaviors:
- SCHEMA: Must precede records for a given stream. It triggers the registration of fields and types. Mid-stream schema updates are supported.
- RECORD: Contains the actual data. Transformation fails if the record's stream has not been previously registered via a SCHEMA message.
- STATE: Signals progress or bookmarks. The target logs these for observability but does not use them for internal data transformation logic.
Sources: src/flext_target_oracle_wms/_utilities/client.py:146-172, tests/unit/test_workflow.py:81-91
Execution Sequence
The following sequence shows how the system handles a standard data stream:
Data Transformation Rules
Values are converted based on the Singer type provided in the schema properties.
| Singer Type | Oracle Target Format | Note |
|---|---|---|
string | str | Default behavior. |
integer | int | Converted via int(str(value)). |
number | float | Converted via float(str(value)). |
boolean | str | Converted to string representation (e.g., "True"). |
object | JSON string | Serialized using UTF-8 encoding. |
array | JSON string | Serialized using UTF-8 encoding. |
Sources: src/flext_target_oracle_wms/_utilities/helpers.py:38-51, tests/unit/test_wms_patterns.py:39-75
Summary
Stream processing in flext-target-oracle-wms provides a robust mechanism for translating Singer data into Oracle-compatible payloads. By decoupling catalog management, table mapping, and data transformation, the system ensures consistent handling of metadata and records across multiple interleaved streams. The enforcement of uppercase keys and strict type conversion ensures compatibility with Oracle WMS Cloud's underlying database structures.