tessl/pypi-ldap3

A strictly RFC 4510 conforming LDAP V3 pure Python client library

1.08x

Overview

Eval results

Files

Abstract Layer

Name: tessl/pypi-ldap3
Rating: 0.81 (1 reviews)
Author: tessl

High-level ORM-like interface with object definitions, attribute definitions, entry objects, and reader classes for simplified LDAP programming.

Capabilities

Object Definition

Define LDAP object schemas with attribute definitions for type-safe and validated entry manipulation.

class ObjectDef:
    def __init__(self, object_class=None):
        """
        Define LDAP object schema.

        Args:
            object_class (str or list, optional): LDAP object class name(s)
        """

    def add(self, definition):
        """
        Add attribute definition to object.

        Args:
            definition (AttrDef): Attribute definition object
        """

    def remove(self, item):
        """
        Remove attribute definition.

        Args:
            item (str or AttrDef): Attribute name or definition to remove
        """

    def clear(self):
        """Clear all attribute definitions."""

    def __iadd__(self, other):
        """Add attribute definition using += operator."""

    def __isub__(self, other):
        """Remove attribute definition using -= operator."""

    def __contains__(self, item):
        """Check if attribute definition exists."""

    def __getitem__(self, item):
        """Get attribute definition by name."""

    def __len__(self):
        """Get number of attribute definitions."""

    def __iter__(self):
        """Iterate over attribute definitions."""

Attribute Definition

Define individual LDAP attributes with validation, transformation, and dereferencing rules.

class AttrDef:
    def __init__(self, name, key=None, validate=None, pre_query=None, post_query=None,
                 default=NotImplemented, dereference_dn=None, description=None):
        """
        Define LDAP attribute with validation and transformation.

        Args:
            name (str): LDAP attribute name
            key (str, optional): Friendly name for Python access
            validate (callable, optional): Validation function for values
            pre_query (callable, optional): Transform values before LDAP query
            post_query (callable, optional): Transform values after LDAP response
            default: Default value if attribute is missing
            dereference_dn (str, optional): DN dereferencing mode
            description (str, optional): Attribute description
        """

Properties:

name: LDAP attribute name
key: Python-friendly name
validate: Validation function
pre_query: Pre-query transformation
post_query: Post-query transformation
default: Default value
dereference_dn: DN dereferencing mode
description: Human-readable description

Attribute Value Container

Container for LDAP attribute values with automatic type conversion and validation.

class Attribute:
    def __init__(self, attr_def, entry):
        """
        Create attribute value container.

        Args:
            attr_def (AttrDef): Attribute definition
            entry (Entry): Parent entry object
        """

    @property
    def value(self):
        """
        Get attribute value(s).

        Returns:
            Single value for single-valued attributes, list for multi-valued
        """

    def __len__(self):
        """Get number of values."""

    def __iter__(self):
        """Iterate over values."""

    def __getitem__(self, item):
        """Get value by index."""

    def __eq__(self, other):
        """Compare attribute values."""

Entry Object

High-level representation of LDAP entries with attribute access and manipulation methods.

class Entry:
    def __init__(self, dn, reader):
        """
        Create entry object.

        Args:
            dn (str): Distinguished Name of entry
            reader (Reader): Reader object that retrieved this entry
        """

    def entry_get_dn(self):
        """
        Get entry distinguished name.

        Returns:
            str: Entry DN
        """

    def entry_get_response(self):
        """
        Get raw LDAP response for entry.

        Returns:
            dict: Raw LDAP response
        """

    def entry_get_reader(self):
        """
        Get reader object that retrieved this entry.

        Returns:
            Reader: Reader object
        """

    def entry_get_raw_attributes(self):
        """
        Get raw attribute values from LDAP response.

        Returns:
            dict: Raw attributes
        """

    def entry_get_raw_attribute(self, name):
        """
        Get raw values for specific attribute.

        Args:
            name (str): Attribute name

        Returns:
            list: Raw attribute values
        """

    def entry_get_attribute_names(self):
        """
        Get list of available attribute names.

        Returns:
            list: Attribute names
        """

    def entry_get_attributes_dict(self):
        """
        Get attributes as dictionary.

        Returns:
            dict: Attribute name to value mapping
        """

    def entry_refresh_from_reader(self):
        """Refresh entry data from reader."""

    def entry_to_json(self):
        """
        Convert entry to JSON format.

        Returns:
            str: Entry in JSON format
        """

    def entry_to_ldif(self):
        """
        Convert entry to LDIF format.

        Returns:
            str: Entry in LDIF format
        """

    def __iter__(self):
        """Iterate over attribute names."""

    def __contains__(self, item):
        """Check if attribute exists in entry."""

    def __getitem__(self, item):
        """Access attribute by name."""

    def __eq__(self, other):
        """Compare entries."""

    def __lt__(self, other):
        """Compare entries for sorting."""

    def entry_get_changes_dict(self):
        """
        Get dictionary of pending changes for entry.

        Returns:
            dict: Dictionary of changes by attribute name
        """

    def entry_commit_changes(self):
        """
        Commit pending changes to LDAP server.

        Returns:
            bool: True if changes committed successfully
        """

    def entry_discard_changes(self):
        """Discard pending changes without committing."""

    def entry_delete(self):
        """
        Delete entry from LDAP server.

        Returns:
            bool: True if entry deleted successfully
        """

    def entry_refresh(self):
        """
        Refresh entry data from LDAP server.

        Returns:
            bool: True if entry refreshed successfully
        """

    def entry_move(self, destination_dn):
        """
        Move entry to new DN location.

        Args:
            destination_dn (str): Destination DN for entry

        Returns:
            bool: True if entry moved successfully
        """

    def entry_rename(self, new_name):
        """
        Rename entry with new relative DN.

        Args:
            new_name (str): New relative DN

        Returns:
            bool: True if entry renamed successfully
        """

Dynamic Attribute Access: Attributes can be accessed as properties:

entry.cn           # Access 'cn' attribute
entry.mail         # Access 'mail' attribute
entry.memberOf     # Access 'memberOf' attribute

Reader Class

High-level search interface with object-oriented query building and result processing.

class Reader:
    def __init__(self, connection, object_def, query, base, components_in_and=True,
                 sub_tree=True, get_operational_attributes=False, controls=None):
        """
        Create reader for high-level LDAP searches.

        Args:
            connection (Connection): LDAP connection object
            object_def (ObjectDef): Object definition for entries
            query (str): Search query string
            base (str): Search base DN
            components_in_and (bool): Combine query components with AND logic
            sub_tree (bool): Use subtree search scope
            get_operational_attributes (bool): Include operational attributes
            controls (list, optional): LDAP controls for search
        """

    def search(self):
        """
        Execute search and populate entries.

        Returns:
            bool: True if search successful
        """

    def search_level(self):
        """
        Execute single-level search.

        Returns:
            bool: True if search successful
        """

    def search_subtree(self):
        """
        Execute subtree search.

        Returns:
            bool: True if search successful
        """

    def search_object(self, entry_dn):
        """
        Search for specific entry by DN.

        Args:
            entry_dn (str): Distinguished Name to search for

        Returns:
            bool: True if search successful
        """

    def search_size_limit(self, size_limit):
        """
        Execute search with size limit.

        Args:
            size_limit (int): Maximum entries to return

        Returns:
            bool: True if search successful
        """

    def search_time_limit(self, time_limit):
        """
        Execute search with time limit.

        Args:
            time_limit (int): Search time limit in seconds

        Returns:
            bool: True if search successful
        """

    def search_types_only(self):
        """
        Execute search returning attribute types only.

        Returns:
            bool: True if search successful
        """

    def search_paged(self, paged_size, paged_criticality=False):
        """
        Execute paged search.

        Args:
            paged_size (int): Page size
            paged_criticality (bool): Paged search criticality

        Returns:
            bool: True if search successful
        """

    def clear(self):
        """Clear search results."""

    def reset(self):
        """Reset reader state."""

    def __iter__(self):
        """Iterate over found entries."""

    def __getitem__(self, item):
        """Get entry by index."""

    def __len__(self):
        """Get number of entries found."""

Properties:

definition: Object definition used
query: Current search query
components_in_and: Query component logic
entries: List of found Entry objects

Operational Attribute

Specialized attribute container for LDAP operational attributes.

class OperationalAttribute(Attribute):
    """
    Operational attribute container (inherits from Attribute).
    
    Handles LDAP operational attributes like createTimestamp, 
    modifyTimestamp, entryUUID, etc.
    """

Usage Examples

Object Definition and Attribute Definition

import ldap3

# Define object schema
person = ldap3.ObjectDef('inetOrgPerson')

# Add attribute definitions with validation
person += ldap3.AttrDef('cn', key='name')
person += ldap3.AttrDef('mail', key='email', validate=lambda x: '@' in x)
person += ldap3.AttrDef('telephoneNumber', key='phone')
person += ldap3.AttrDef('employeeNumber', key='emp_id', 
                       validate=lambda x: x.isdigit())

# Add attribute with default value
person += ldap3.AttrDef('department', default='IT')

Reader-based Searching

# Create connection and reader
server = ldap3.Server('ldap://ldap.example.com')
conn = ldap3.Connection(server, 'cn=user,dc=example,dc=com', 'password', auto_bind=True)

# Create reader with object definition
reader = ldap3.Reader(conn, person, 'name: John*', 'ou=people,dc=example,dc=com')

# Perform search
reader.search()

# Access results through entry objects
for entry in reader:
    print(f"Name: {entry.name}")           # Using key from AttrDef
    print(f"Email: {entry.email}")         # Validated email attribute
    print(f"Phone: {entry.phone}")
    print(f"Employee ID: {entry.emp_id}")  # Validated numeric
    print(f"Department: {entry.department}") # May use default value
    print(f"DN: {entry.entry_get_dn()}")
    print("---")

Advanced Query Building

# Complex queries using Reader
reader = ldap3.Reader(conn, person, 
                     'name: John* & email: *@company.com | department: Engineering',
                     'dc=example,dc=com')

# Execute subtree search
reader.search_subtree()

# Access entry details
for entry in reader:
    # Get all attributes as dictionary
    attrs = entry.entry_get_attributes_dict()
    print(f"All attributes: {attrs}")
    
    # Convert to different formats
    json_data = entry.entry_to_json()
    ldif_data = entry.entry_to_ldif()

Working with Multi-valued Attributes

# Define group object with member attribute
group = ldap3.ObjectDef('groupOfNames')
group += ldap3.AttrDef('cn', key='name')
group += ldap3.AttrDef('member', key='members')

reader = ldap3.Reader(conn, group, 'name: *Admin*', 'ou=groups,dc=example,dc=com')
reader.search()

for group_entry in reader:
    print(f"Group: {group_entry.name}")
    
    # Multi-valued attribute access
    if hasattr(group_entry, 'members'):
        print(f"Number of members: {len(group_entry.members)}")
        for member in group_entry.members:
            print(f"  Member: {member}")

Custom Validation and Transformation

import re
from datetime import datetime

def validate_email(email):
    """Email validation function."""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def format_phone(phone):
    """Phone number formatting function."""
    # Remove non-digits
    digits = ''.join(filter(str.isdigit, phone))
    if len(digits) == 10:
        return f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"
    return phone

def parse_timestamp(timestamp):
    """Parse LDAP timestamp to datetime object."""
    if isinstance(timestamp, list):
        timestamp = timestamp[0]
    return datetime.strptime(timestamp, '%Y%m%d%H%M%SZ')

# Create object definition with custom functions
person = ldap3.ObjectDef('inetOrgPerson')
person += ldap3.AttrDef('cn', key='name')
person += ldap3.AttrDef('mail', key='email', validate=validate_email)
person += ldap3.AttrDef('telephoneNumber', key='phone', post_query=format_phone)
person += ldap3.AttrDef('createTimestamp', key='created', post_query=parse_timestamp)

reader = ldap3.Reader(conn, person, 'name: *', 'ou=people,dc=example,dc=com', 
                     get_operational_attributes=True)
reader.search()

for entry in reader:
    print(f"Name: {entry.name}")
    print(f"Email: {entry.email}")        # Validated
    print(f"Phone: {entry.phone}")        # Formatted
    print(f"Created: {entry.created}")    # Parsed datetime

Pagination with Reader

# Reader-based paged search
reader = ldap3.Reader(conn, person, 'name: *', 'dc=example,dc=com')

# Search with pagination
page_size = 100
reader.search_paged(page_size)

print(f"Found {len(reader)} entries in first page")

# Continue with more pages if needed
while True:
    # Check if more results available
    if not conn.result.get('controls', {}).get('1.2.840.113556.1.4.319', {}).get('value', {}).get('cookie'):
        break
    
    reader.search_paged(page_size)
    print(f"Found {len(reader)} additional entries")

DN Dereferencing

# Dereference DN attributes to get referenced entry data
person = ldap3.ObjectDef('inetOrgPerson')
person += ldap3.AttrDef('cn', key='name')
person += ldap3.AttrDef('manager', key='manager_info', dereference_dn='cn')

reader = ldap3.Reader(conn, person, 'name: *', 'ou=people,dc=example,dc=com')
reader.search()

for entry in reader:
    print(f"Employee: {entry.name}")
    if hasattr(entry, 'manager_info'):
        print(f"Manager: {entry.manager_info}")  # Dereferenced manager CN

Error Handling in Abstract Layer

try:
    # Create object definition
    person = ldap3.ObjectDef('inetOrgPerson')
    person += ldap3.AttrDef('mail', validate=lambda x: '@' in x)
    
    reader = ldap3.Reader(conn, person, 'mail: *', 'dc=example,dc=com')
    reader.search()
    
    for entry in reader:
        # Access validated attributes
        print(f"Valid email: {entry.mail}")
        
except ldap3.LDAPAttributeError as e:
    print(f"Attribute error: {e}")
except ldap3.LDAPEntryError as e:
    print(f"Entry error: {e}")
except ldap3.LDAPReaderError as e:
    print(f"Reader error: {e}")

Combining Abstract Layer with Raw Operations

# Use abstract layer for searching, raw operations for modifications
reader = ldap3.Reader(conn, person, 'name: John*', 'ou=people,dc=example,dc=com')
reader.search()

for entry in reader:
    # Get DN from abstract entry
    entry_dn = entry.entry_get_dn()
    
    # Use raw connection for modifications
    changes = {'description': [(ldap3.MODIFY_REPLACE, f'Updated by script on {datetime.now()}')]}
    conn.modify(entry_dn, changes)
    
    print(f"Updated entry: {entry.name}")

Install with Tessl CLI

npx tessl i tessl/pypi-ldap3