Stream Processor Architecture
Relevant source files
The following files were used as context for generating this wiki page:
Introduction
The Stream Processor Architecture in flext-tap-ldif provides a robust framework for extracting LDAP Data Interchange Format (LDIF) data and converting it into Singer-compatible streams. This system leverages the flext-ldif infrastructure to parse files, validate content, and map LDIF entries to structured JSON records suitable for ETL pipelines managed by Meltano or other Singer targets.
The architecture is primarily centered around the Processor and EntriesStream classes, which coordinate file discovery, content parsing, and record generation. By utilizing MRO mixins and Pydantic models, the system ensures strict configuration validation and high performance during batch processing of large LDIF files.
Sources: src/flext_tap_ldif/_utilities/processor.py, src/flext_tap_ldif/_utilities/entries_stream.py:12-16, src/flext_tap_ldif/tap.py:16-18
Core Components
The architecture consists of three primary layers: the Tap entry point, the Stream management layer, and the Processor utility layer.
1. Processor (FlextTapLdifUtilitiesProcessor.Processor)
The Processor is the engine of the architecture. It interacts directly with the flext-ldif API to handle the low-level parsing of LDIF content. It is responsible for:
- Discovery: Locating LDIF files based on directory paths or specific file patterns.
- Parsing: Converting raw file content into
flext_ldifentry models. - Record Mapping: Transforming raw entries into the final Singer record format, including metadata like source file paths and entry sizes.
Sources: src/flext_tap_ldif/_utilities/processor.py:22-30
2. Entries Stream (FlextTapLdifUtilitiesEntriesStream.EntriesStream)
This component inherits from Meltano.SingerStreamBase and acts as the bridge between the Singer protocol and the LDIF processor. It defines the JSON schema for the output data and manages the iteration over discovered files to yield SingerRecord objects.
Sources: src/flext_tap_ldif/_utilities/entries_stream.py:16-25
3. Data Processing Utilities (FlextTapLdifUtilitiesLdifDataProcessing)
A specialized utility class that provides static methods for normalizing attribute names and handling complex LDIF line formats, including Base64 encoded values and URL references.
Sources: src/flext_tap_ldif/_utilities/data_processing.py:13-17
Data Flow and Logic
The following diagram illustrates the sequence of operations from the initial tap discovery to the final record generation.
File Discovery Logic
The discover_files method supports both single file targeting and directory-based globbing. It enforces a safety limit via max_file_size_mb to prevent memory exhaustion when reading large files.
| Parameter | Type | Description |
|---|---|---|
directory_path | str | Path | The directory to scan for LDIF files. |
file_pattern | str | Glob pattern (defaults to *.ldif). |
file_path | str | Path | Direct path to a specific file. |
max_file_size_mb | int | Maximum allowed file size for processing. |
Sources: src/flext_tap_ldif/_utilities/processor.py:32-63
Data Models and Schemas
Stream Configuration Model
The LdifStream model (an EnforcedModel) validates the configuration used by the processor.
| Field | Type | Description | Default |
|---|---|---|---|
stream_name | str | Name of the Singer stream. | Required |
file_path | str | Path to the source LDIF file. | Required |
replication_method | str | Singer replication strategy. | FULL_TABLE |
batch_size | int | Number of entries per processing batch. | 1000 |
include_change_records | bool | Whether to extract LDIF change records. | True |
Sources: src/flext_tap_ldif/_models/file_stream.py:18-50
Output Record Schema
The system generates records with a fixed schema defined in the EntriesStream. This ensures that downstream targets receive consistent data structures regardless of the source LDIF's complexity.
# Schema definition for LDIF entries
{
"dn": {"type": "string"},
"attributes": {"type": "object"},
"object_class": {"type": "array", "items": {"type": "string"}},
"change_type": {"type": "string"},
"source_file": {"type": "string"},
"line_number": {"type": "integer"},
"entry_size": {"type": "integer"}
}
Sources: src/flext_tap_ldif/_utilities/entries_stream.py:104-118, src/flext_tap_ldif/tap.py:58-69
Processing Lifecycle
The processor handles file reading and entry extraction through a generator-based lifecycle to maintain a low memory footprint.
Attribute Normalization
LDIF attributes often contain characters (like hyphens) or naming conventions incompatible with standard JSON schema expectations. The normalize_ldif_attribute_name method ensures:
- All names are lowercased.
- Non-alphanumeric characters are replaced with underscores.
- Names starting with digits are prefixed with
attr_.
Sources: src/flext_tap_ldif/_utilities/data_processing.py:81-91
Error Handling and Strictness
The architecture supports a strict_parsing configuration. If enabled (default), the processor will raise exceptions on parsing failures, halting the stream. If disabled, it logs failures and continues to the next file or entry, ensuring resilience in multi-file extraction scenarios.
Sources: src/flext_tap_ldif/_utilities/processor.py:77-83, src/flext_tap_ldif/_utilities/entries_stream.py:84-95
Summary
The Stream Processor Architecture provides a highly structured and validated pipeline for LDIF data. By decoupling file discovery and parsing into a dedicated Processor and leveraging Pydantic for configuration validation via the LdifStream model, the system achieves a clean separation of concerns. This architecture ensures that complex LDIF attributes are consistently normalized and mapped to the Singer specification, facilitating seamless integration with the wider FLEXT ecosystem.