End-to-End Workflows
Relevant source files
The following files were used as context for generating this wiki page:
End-to-end workflows in the flext-target-oracle project define the complete lifecycle of data ingestion from Singer-formatted streams into an Oracle database. This process encompasses configuration validation, schema discovery, table creation, record batching, and state management. The system utilizes a coordinate structure involving the FlextTargetOracle client and the FlextTargetOracleLoader to ensure that data flows from JSON payloads to persistent Oracle tables with transactional integrity.
The workflows are designed to handle various message types defined by the Singer spec, primarily SCHEMA, RECORD, and STATE. The implementation follows a "Railway-Oriented Programming" pattern, utilizing result objects to handle errors gracefully across the pipeline.
System Architecture and Components
The end-to-end workflow is orchestrated by two primary components that manage the transition from raw Singer messages to database operations.
Key Components
| Component | Description | Primary File |
|---|---|---|
| FlextTargetOracle | The main client interface that parses Singer payloads and routes messages to appropriate handlers. | src/flext_target_oracle/_utilities/client.py |
| FlextTargetOracleLoader | The data access layer responsible for DDL (table creation/updates) and DML (inserting/merging records). | src/flext_target_oracle/_utilities/loader.py |
| FlextDbOracleApi | An external API used by the loader for actual SQL execution and connection management. | src/flext_target_oracle/_utilities/loader.py |
Sources: src/flext_target_oracle/_utilities/client.py:20-28, src/flext_target_oracle/_utilities/loader.py:25-35
Data Flow Overview
The following diagram illustrates how a Singer message moves through the system from ingestion to the Oracle database.
The workflow ensures that tables are prepared before any records are processed and that records are efficiently batched based on configuration.
Sources: src/flext_target_oracle/_utilities/client.py:160-172, src/flext_target_oracle/_utilities/loader.py:380-405
The Singer Lifecycle Workflow
A typical end-to-end execution follows a strict sequence of message types to ensure data integrity and schema synchronization.
1. Initialization and Schema Handling
The workflow begins with the initialize() method, which verifies connectivity to the Oracle instance. Upon receiving a SCHEMA message, the target builds Oracle column metadata, including primary keys and audit columns (e.g., _SDC_EXTRACTED_AT).
- Column Mapping: The system supports renaming source fields to target Oracle columns and filtering ignored columns.
- Table Creation: If the table does not exist, the
FlextTargetOracleLoadergenerates and executes the necessary DDL.
Sources: src/flext_target_oracle/_utilities/client.py:155-158, src/flext_target_oracle/_utilities/loader.py:95-115, tests/integration/test_oracle.py:305-320
2. Record Processing and Batching
When a RECORD message is received, it is not immediately written to the database. Instead, it is transformed into a set of parameters and placed in an internal buffer.
- Batch Size: Records are flushed to the database when the buffer size reaches the configured
batch_size. - Transformation: Nested JSON fields (arrays/objects) are handled based on the
storage_mode. Injsonorhybridmode, they are serialized to CLOB columns. - Audit Fields: Every record is enriched with
_SDC_LOADED_ATtimestamps during the flush operation.
Sources: src/flext_target_oracle/_utilities/loader.py:408-418, src/flext_target_oracle/_utilities/loader.py:488-510, tests/integration/test_oracle.py:180-210
3. State and Finalization
The STATE message updates the target's internal bookmarking, allowing for incremental syncs. The workflow concludes with the finalize() call, which flushes any remaining records in the buffers to the database and returns a summary of the operation.
Sources: src/flext_target_oracle/_utilities/client.py:228-232, src/flext_target_oracle/_utilities/client.py:148-150
Production and Enterprise Configuration
In production environments, the workflow is often wrapped in a manager that handles signals, health checks, and environment-based configuration.
The production setup ensures that resources are cleaned up via a graceful shutdown process that handles SIGINT and SIGTERM signals.
Sources: examples/03_production_setup.py:126-150, examples/03_production_setup.py:290-305
Performance Optimization Strategies
End-to-end workflows support several loading methods to optimize for different data volumes and requirements:
| Load Method | Description |
|---|---|
| Bulk Insert | Uses Oracle bulk operations to insert large datasets efficiently. |
| Merge | Deletes existing rows based on primary keys before inserting new ones to simulate UPSERT. |
| Truncate | Clears the target table entirely before loading the new batch. |
Sources: tests/integration/test_oracle.py:155-175, src/flext_target_oracle/_utilities/loader.py:545-565, examples/01_basic_usage.py:55-65
Conclusion
The End-to-End Workflow in flext-target-oracle provides a robust, batched approach to data synchronization. By separating message parsing (the client) from database interaction (the loader), the system maintains high performance through bulk operations while ensuring that schema evolution and state tracking are handled according to the Singer specification. This structure supports both simple local scripts and complex enterprise-grade production deployments.