LDAP Synchronization Service
Relevant source files
The following files were used as context for generating this wiki page:
The LDAP Synchronization Service is a core component of the flext-dbt-ldap project, designed to facilitate the extraction of directory data from LDAP servers and its subsequent transformation and synchronization into a data warehouse using dbt (data build tool). It provides a structured workflow for handling users, groups, and memberships, supporting both full and incremental synchronization modes to ensure data consistency and freshness.
The service is implemented primarily through a mixin-based architecture, notably FlextDbtLdapSyncMixin and FlextDbtLdapClientMixin, which are composed into the unified FlextDbtLdap facade. This architecture allows the service to leverage lower-level LDAP client capabilities while providing high-level synchronization logic, state management via bookmarks, and data quality validation.
Sources: src/flext_dbt_ldap/api.py:22-38, src/flext_dbt_ldap/services/sync.py:16-20
Service Architecture
The service follows a layered architecture where the public API (FlextDbtLdap) inherits functionality from specialized mixins. The FlextDbtLdapSyncMixin handles the high-level orchestration of sync tasks, while FlextDbtLdapClientMixin manages the direct interactions with the LDAP server and the execution of dbt models.
Component Relationship
The following diagram illustrates the relationship between the primary service components and the external systems (LDAP and Warehouse).
Synchronization Workflows
The service provides three primary synchronization methods targeting specific LDAP entities, as well as a "full" sync that orchestrates all of them in sequence.
Supported Sync Components
| Component | Description | DBT Models Involved |
|---|---|---|
| Users | Syncs LDAP user entries (e.g., inetOrgPerson) | stg_users, dim_users |
| Groups | Syncs LDAP group entries (e.g., groupOfNames) | stg_groups, dim_groups |
| Memberships | Syncs relationships between users and groups | fact_memberships |
Sources: src/flext_dbt_ldap/services/sync.py:84-187
Full Data Warehouse Sync
The run_full_data_warehouse_sync method executes a comprehensive update. It sequentially calls user sync, group sync, and membership sync, aggregating the results into a SyncResult model.
Incremental Sync and State Management
To optimize performance and reduce load on the LDAP server, the service supports incremental synchronization using LDAP modifyTimestamp bookmarks.
Bookmark Logic
The service maintains a local state file (typically .flext_dbt_ldap_sync_state.json) which stores the last successful sync timestamp for each entity type.
- Check: Before a sync, the service checks if
incremental=Truewas requested. - Filter: If an existing bookmark is found, it constructs an LDAP filter extension:
(&{base_filter}(modifyTimestamp>={bookmark})). - Update: Upon successful completion of the dbt pipeline, the bookmark is updated to the current UTC time.
Sources: src/flext_dbt_ldap/services/sync.py:189-215, src/flext_dbt_ldap/services/sync.py:246-277
State Persistence
State is persisted to disk using JSON. If the persistence fails (e.g., disk full), the sync operation returns a failure to prevent state divergence.
# From src/flext_dbt_ldap/services/sync.py
def _update_bookmark(self, sync_key: str, bookmark: str, *, successful: bool) -> p.Result[bool]:
if not successful:
return r[bool].ok(True)
previous_bookmark = self._sync_bookmarks.get(sync_key)
self._sync_bookmarks[sync_key] = bookmark
try:
self._persist_sync_state()
except OSError as error:
# Rollback in-memory state on persistence failure
self._sync_bookmarks[sync_key] = previous_bookmark
return r[bool].fail(str(error))
return r[bool].ok(True)
Sources: src/flext_dbt_ldap/services/sync.py:261-277
Data Extraction and Transformation
The underlying FlextDbtLdapClientMixin handles the transition of data from LDAP attributes to dbt-compatible formats.
Pipeline Execution Flow
The run_full_pipeline method defines the standard lifecycle for moving data:
- Extraction: Uses the
_ldap_apito search for entries based on the provided filter and attributes. - Validation: Checks extracted entries against required attributes and a minimum quality threshold (configured in settings).
- Transformation: Executes dbt models to process the data in the warehouse.
Attribute Mapping
The service maps LDAP attributes to warehouse column names based on the ldap_attribute_mapping configuration. It also handles the parsing of objectClass to identify entries as users, groups, or organizational units.
Sources: src/flext_dbt_ldap/services/client.py:214-230, src/flext_dbt_ldap/services/client.py:232-243
Result and Metric Models
The service uses structured Pydantic models to return execution results and quality metrics.
| Model | Field | Description |
|---|---|---|
ValidationMetrics | quality_score | Ratio of valid entries to total entries. |
validation_passed | Boolean indicating if the score meets the threshold. | |
DbtRunStatus | models_run | Sequence of model names executed. |
entries_processed | Count of LDAP records handled. | |
SyncResult | overall_success | True only if all sub-components succeeded. |
Sources: src/flext_dbt_ldap/_models/results.py:15-62
Summary
The LDAP Synchronization Service provides a robust, stateful bridge between enterprise directory services and modern data warehouses. By combining LDAP-native features like modifyTimestamp with dbt's transformation capabilities, it enables efficient data pipelines. Key features include automated bookmarking for incremental updates, configurable attribute mapping, and rigorous data quality validation to ensure that only compliant directory data reaches the warehouse analytics layer.
Sources: src/flext_dbt_ldap/services/sync.py:100-180, src/flext_dbt_ldap/services/client.py:168-202