Oracle WMS Streams Catalog

Relevant source files

The following files were used as context for generating this wiki page:

• src/flext_tap_oracle_wms/streams.py
• src/flext_tap_oracle_wms/tap.py
• tests/integration/test_streams_functional.py
• tests/integration/test_functional.py
• tests/e2e/test_e2e.py
• meltano.yml

The Oracle WMS Streams Catalog system is responsible for discovering, defining, and managing data streams extracted from the Oracle Warehouse Management System (WMS). It acts as the bridge between the Oracle WMS REST API and the Singer protocol, converting WMS entities into a structured Singer catalog. This system ensures that complex warehouse data—such as inventory, locations, and orders—is mapped to valid JSON schemas for downstream consumption.

Sources: src/flext_tap_oracle_wms/tap.py:14-25, src/flext_tap_oracle_wms/streams.py:20-25, meltano.yml:195-215

Stream Architecture and Lifecycle

The streams are implemented via the FlextTapOracleWmsStream class, which extends the RESTStream patterns provided by the Singer SDK. Each stream is dynamic and adapts to specific Oracle WMS entities based on discovery metadata.

The diagram shows the lifecycle from entity discovery to the generation of Singer-compatible records.
Sources: src/flext_tap_oracle_wms/tap.py:180-205, src/flext_tap_oracle_wms/streams.py:27-40

Key Stream Components

Component	Description	Source
discover_streams	Orchestrates the discovery of available WMS entities and builds stream objects.	tap.py:207-226
get_records	The primary data extraction loop that handles pagination and API requests.	streams.py:118-140
post_process	Handles record-level transformations, including column mapping and ignoring sensitive fields.	streams.py:145-190
_fetch_page_data	Interacts with the wms_client to retrieve a specific page of entity data.	streams.py:208-238

Data Discovery and Schema Generation

The discovery process identifies entities available in the target Oracle WMS instance. The system automatically detects replication methods and identifies primary keys to populate the Singer Catalog.

Automated Discovery Flow

This flow illustrates how the tap queries the WMS metadata to build a complete catalog.
Sources: src/flext_tap_oracle_wms/tap.py:175-185, tests/integration/test_functional.py:82-110

Replication Methods

The catalog defines how data is updated over time. Two primary methods are supported:

1. FULL_TABLE: Entire entities are synced (e.g., company, facility, location).
2. INCREMENTAL: Only new or modified records are synced using a replication-key such as last_updated_ts or mod_ts.

Sources: meltano.yml:218-235, tests/integration/test_functional.py:218-245

Data Normalization and Transformation

Oracle WMS returns data in formats that may not directly comply with the Singer JSON specification. The FlextTapOracleWmsStream includes normalization logic to ensure data integrity.

Normalization Logic

• Scalar Normalization: Converts non-JSON runtime scalars (like Path or custom BaseModel objects) into strings.
• Complex Object Mapping: Recursively processes lists and maps using the FlextOracleWmsUtilities.MappingConversion tool.
• Column Mapping: Renames fields based on tap configuration (e.g., mapping internal WMS codes to human-readable names).
• Field Exclusion: Automatically removes fields specified in the ignored_columns configuration.

Sources: src/flext_tap_oracle_wms/streams.py:84-114, src/flext_tap_oracle_wms/streams.py:145-185

Configuration and Pagination

The catalog behavior is influenced by several settings defined in the tap's schema and environment.

Setting	Type	Default	Description
page_size	Integer	100	Number of records to fetch per API request (Max 1250).
pagination_mode	Option	offset	Strategy for navigating large datasets (offset or cursor).
base_url	String	N/A	The root URL of the Oracle WMS instance.
api_version	String	v1	Target version of the Oracle WMS REST API.

Sources: meltano.yml:105-125, src/flext_tap_oracle_wms/tap.py:30-55, src/flext_tap_oracle_wms/streams.py:65-75

Pagination Logic

The stream implements HATEOAS-aware pagination. The _build_operation_kwargs method prepares the request by calculating the correct limit and applying filter parameters for incremental syncs. A stream continues fetching as long as the record count of the returned page matches the requested page_size.

Sources: src/flext_tap_oracle_wms/streams.py:192-206, src/flext_tap_oracle_wms/streams.py:233-238

Stream Implementation Example

The following snippet demonstrates how a stream is initialized and how it manages the WMS client connection.

# From src/flext_tap_oracle_wms/streams.py:42-82
class FlextTapOracleWmsStream(m.Meltano.SingerStreamBase):
    def __init__(self, tap, name=None, schema=None):
        # ... validation and initialization ...
        self._client = None
        self._page_size = self._validate_page_size(tap.settings.get("page_size"))

    @property
    def client(self) -> FlextOracleWmsUtilities.OracleWms.Client:
        if self._client is not None:
            return self._client
        # Retrieves the shared client from the tap instance
        self._client = self._tap.wms_client
        return self._client

Sources: src/flext_tap_oracle_wms/streams.py:42-82

Summary

The Oracle WMS Streams Catalog provides a robust, dynamic framework for data extraction. By automating entity discovery and enforcing strict Singer-compliant normalization, it ensures that warehouse data is reliably available for integration. Its support for both Full Table and Incremental replication allows for efficient data synchronization tailored to the specific volatility of warehouse entities like inventory and shipments.