The Processing Pipeline
Relevant source files
The following files were used as context for generating this wiki page:
The Processing Pipeline in flext-ldif is a core service-layer orchestration component designed to handle end-to-end LDIF data transformations. It operates as a sequence of stages that normalize, validate, and transform LDIF entries, moving them from a source server format to a target server format. The pipeline is built on the Railway Oriented Programming pattern, ensuring that failures in any stage result in early detection and consistent error reporting.
Architecturally, the pipeline resides in the Infrastructure Layer as a domain service, orchestrating low-level LDIF processing while providing a high-level API for application use cases. It supports batch processing, context-aware execution with correlation tracking, and integration with the flext-cli pipeline specifications.
Sources: src/README.md:33-45, src/flext_ldif/services/pipeline.py:12-25, examples/01_basic_usage.py:100-112
Pipeline Architecture and Components
The pipeline is implemented primarily through the FlextLdifProcessingPipeline class. It manages a batch of m.Ldif.Entry objects and applies a series of FlextLdifTransformer instances configured via a TransformConfig.
Execution Flow
The execution logic follows a linear sequence where entry batches are passed through stage spec handlers. If the pipeline is configured for server migration, it prioritizes server-specific transformations before applying general normalizations.
The diagram above shows the logical flow of _build_pipeline and execute methods.
Sources: src/flext_ldif/services/pipeline.py:75-103, src/flext_ldif/services/pipeline.py:133-207
Key Classes and Data Structures
| Component | Responsibility | Source |
|---|---|---|
FlextLdifProcessingPipeline | Orchestrates the execution of multiple processing stages. | pipeline.py:12 |
FlextLdifMigrationPipeline | Handles directory-level migration and file I/O for pipelines. | migration.py:15 |
m.Ldif.TransformConfig | Configuration defining which normalization and transformation steps to run. | pipeline.py:27 |
FlextLdifTransformer | Service implementing specific server-to-server transformation logic. | pipeline.py:186 |
p.Result | A railway-oriented container for success values or failure errors. | pipeline.py:77 |
Processing Stages
The pipeline builds its execution path dynamically based on the configuration provided.
DN Normalization
This stage utilizes u.Ldif.Normalize.dn to ensure Distinguished Names (DNs) adhere to specific case folding and space handling rules.
- Case Fold Options:
c.Ldif.CaseFoldOption(NONE, LOWER, UPPER). - Space Handling:
c.Ldif.SpaceHandlingOption(PRESERVE, TRIM).
Sources: src/flext_ldif/services/pipeline.py:137-160
Attribute Normalization
This stage applies u.Ldif.Normalize.attrs to the entries. It can be configured to:
- Trim whitespace from attribute values.
- Remove empty attributes.
- Fold the case of attribute names.
Sources: src/flext_ldif/services/pipeline.py:162-178
Server Transformation
When migrating between server types (e.g., OID to OUD), the FlextLdifTransformer is injected. This is typically the first stage in the pipeline to ensure server-specific logic is applied before normalization.
Sources: src/flext_ldif/services/pipeline.py:179-206
Migration Integration
The FlextLdifMigrationPipeline extends the processing pipeline concepts to handle filesystem operations, such as reading from an input_dir and writing to an output_dir.
This sequence illustrates the integration between migration management and the processing pipeline.
Sources: src/flext_ldif/services/migration.py:112-140, src/flext_ldif/services/migration.py:148-180
Configuration Options
The pipeline behavior is controlled by m.Ldif.TransformConfig and m.Ldif.ProcessConfig.
| Parameter | Type | Default | Description |
|---|---|---|---|
normalize_dns | bool | False | Enables DN normalization stage. |
normalize_attrs | bool | False | Enables Attribute normalization stage. |
source_server | `str | ServerTypes` | RFC |
target_server | `str | ServerTypes` | RFC |
base_dn | str | "" | Base DN used for scope filtering during transformation. |
Sources: src/flext_ldif/services/pipeline.py:133-207, src/flext_ldif/services/migration.py:17-64
Implementation Example: Batch Transformation
The pipeline supports direct batch transformation of entry lists, commonly used in testing and advanced library workflows.
@classmethod
def batch_transform(cls) -> p.Result[list[m.Ldif.Entry]]:
"""DRY batch transformation - returns created entries."""
entries: list[m.Ldif.Entry] = [cls._build_entry(index) for index in range(10)]
return ldif.validate_entries(entries).map(lambda _: entries)
Sources: examples/01_basic_usage.py:95-98
Summary
The Processing Pipeline provides a structured, multi-stage mechanism for LDIF data manipulation. By leveraging a stage-based spec and the Railway Oriented Programming pattern, it ensures that complex LDAP migrations—involving DN normalization, attribute cleanup, and server-specific schema mapping—are performed reliably and transparently. The system is designed for high performance, supporting batch operations and integration with standard CLI pipeline execution models.
Sources: src/README.md:85-88, src/flext_ldif/services/pipeline.py:100-112