File Streaming

Relevant source files

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

• src/flext_tap_ldif/_models/file_stream.py
• src/flext_tap_ldif/_utilities/entries_stream.py
• src/flext_tap_ldif/_utilities/processor.py
• src/flext_tap_ldif/_models/file_metadata.py
• src/flext_tap_ldif/_utilities/state_management.py
• src/flext_tap_ldif/tap.py

The File Streaming system in flext-tap-ldif provides a robust mechanism for discovering, parsing, and yielding records from LDIF (LDAP Data Interchange Format) files. It acts as the core engine for the Singer tap, transforming raw file content into structured Singer records. The system is designed to handle both single file paths and directory-based discovery with pattern matching.

The architecture relies on the EntriesStream class, which orchestrates the flow between the Singer tap interface and the underlying Processor utility. It manages file-level metadata, ensures format validity, and supports state-aware processing to track progress across large datasets.

Core Architecture and Data Flow

The streaming process follows a structured sequence: configuration ingestion, file discovery, record parsing via the flext-ldif API, and finally, record emission to the Singer target.

Operational Sequence

1. Configuration: The stream is initialized with tap settings, including file paths, patterns, and encoding preferences.
2. Discovery: The Processor scans the local filesystem for matching .ldif files that meet size constraints.
3. Parsing: Files are read and parsed into LDIF entries, extracting Distinguished Names (DNs) and attribute mappings.
4. Emission: Entries are encapsulated into SingerRecord objects and yielded to the stream.

The following diagram illustrates the interaction between the tap, the stream manager, and the file processor:

Key Components

LdifStream Model

The LdifStream class defines the configuration schema for the stream. It validates mandatory fields like stream_name and file_path while providing defaults for replication methods and batch sizes.

Field	Type	Default	Description
stream_name	str	N/A	Singer stream name
file_path	str	N/A	Path to the source LDIF file
replication_method	str	FULL_TABLE	Sync strategy
batch_size	int	10000	Processing batch size
include_change_records	bool	True	Whether to include LDIF change records

Sources: src/flext_tap_ldif/_models/file_stream.py:17-68

Processor Utility

The Processor is the engine responsible for filesystem interaction and LDIF parsing. It uses the flext-ldif API to convert raw text into structured models and performs size-based filtering to prevent memory exhaustion.

• discover_files: Validates paths and filters files based on max_file_size_mb.
• process_file: Reads file content using the specified encoding and yields parsed JSON mappings.
• _yield_records: Maps raw LDIF entries to the internal Singer schema, including source metadata like line numbers and file paths.

Sources: src/flext_tap_ldif/_utilities/processor.py:21-125

State and Progress Tracking

To support large file processing, the system includes a StateManagement utility. This allows the tap to resume from the last processed position in a file if interrupted.

The state is maintained as a mapping of file paths to their respective processing offsets.

Sources: src/flext_tap_ldif/_utilities/state_management.py:34-80

Data Schema

The EntriesStream produces records with a standardized schema to ensure consistency across various LDAP exports.

Property	Type	Description
dn	string	The Distinguished Name of the LDAP entry
attributes	object	Key-value pairs of LDAP attributes
object_class	array	List of object classes associated with the entry
source_file	string	Path of the file the record was extracted from
line_number	integer	Starting line number of the entry in the source file

Sources: src/flext_tap_ldif/_utilities/entries_stream.py:116-130, src/flext_tap_ldif/tap.py:53-65

Metadata and Validation

The system tracks processing health through the LdifFile and LdifValidationResult models. These models store counts for entries, change records, and comment lines, alongside any validation errors encountered during parsing.

# Example processing summary computed by the model
{
    "file_path": "/data/users.ldif",
    "file_stats": {
        "size_bytes": 1048576,
        "total_lines": 5000,
        "entries": 450,
        "change_records": 0
    },
    "processing": {
        "status": "completed",
        "progress_percent": 100.0,
        "errors": 0
    }
}

Sources: src/flext_tap_ldif/_models/file_metadata.py:73-95, src/flext_tap_ldif/_models/record.py:64-90

Conclusion

File Streaming in flext-tap-ldif provides a decoupled architecture where the EntriesStream manages the Singer protocol, while the Processor and StateManagement utilities handle the technical complexities of LDIF parsing and incremental syncs. This structure ensures that the tap remains performant and reliable even when dealing with large-scale LDAP data exports.