tessl/pypi-bonsai
Python 3 module for accessing LDAP directory servers with async framework support and Active Directory integration.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
ldif-support.mddocs/
LDIF Format Support
Read and write LDAP Data Interchange Format (LDIF) files for data import/export, backup operations, and batch processing. LDIF support includes parsing, writing, change records, and URL loading capabilities.
Capabilities
LDIF Reader
Parse LDIF files into LDAP entries with support for continuation lines, comments, and external URL references.
from bonsai.ldif import LDIFReader, LDIFError
class LDIFReader:
def __init__(
self,
input_file: TextIO,
autoload: bool = True,
max_length: int = 76
) -> None:
"""
Initialize LDIF reader for parsing LDIF format files.
Parameters:
- input_file: File-like object in text mode containing LDIF data
- autoload: Allow automatic loading of external URL sources
- max_length: Maximum line length allowed in LDIF file
"""
def parse(self) -> Iterator[Tuple[str, Dict[str, List[Union[str, bytes]]]]]:
"""
Parse LDIF file and yield entries.
Yields:
Tuple of (dn, attributes_dict) for each entry
Where attributes_dict maps attribute names to lists of values.
Binary values are returned as bytes, text values as strings.
"""
def parse_entry_records(self) -> Iterator[LDAPEntry]:
"""
Parse LDIF file and yield LDAPEntry objects.
Yields:
LDAPEntry objects for each entry in the LDIF file
"""
def parse_change_records(self) -> Iterator[Tuple[str, str, Dict[str, Any]]]:
"""
Parse LDIF change records (add, modify, delete, modrdn operations).
Yields:
Tuple of (dn, changetype, change_data) for each change record
Where:
- dn: Distinguished name of the entry
- changetype: Operation type ("add", "modify", "delete", "modrdn")
- change_data: Dictionary with operation-specific data
"""
def parse_entries_and_change_records(self) -> Iterator[Union[
Tuple[str, Dict[str, List[Union[str, bytes]]]],
Tuple[str, str, Dict[str, Any]]
]]:
"""
Parse LDIF file yielding both entries and change records.
Yields:
Mixed entries and change records as tuples
"""
@property
def version(self) -> Optional[int]:
"""LDIF version number from version line."""
@property
def autoload(self) -> bool:
"""Whether automatic URL loading is enabled."""
@property
def max_length(self) -> int:
"""Maximum allowed line length."""
def set_resource_handler(self, scheme: str, handler: Callable[[str], bytes]) -> None:
"""
Set custom handler for loading external resources by URL scheme.
Parameters:
- scheme: URL scheme (e.g., "http", "ftp")
- handler: Function that takes URL string and returns bytes
"""
def get_resource_handler(self, scheme: str) -> Optional[Callable[[str], bytes]]:
"""Get resource handler for URL scheme."""LDIF Writer
Write LDAP entries and change records to LDIF format files with proper encoding and formatting.
from bonsai.ldif import LDIFWriter
class LDIFWriter:
def __init__(
self,
output_file: TextIO,
base64_attrs: Optional[List[str]] = None,
cols: int = 76
) -> None:
"""
Initialize LDIF writer for creating LDIF format files.
Parameters:
- output_file: File-like object in text mode for writing LDIF data
- base64_attrs: List of attribute names to always base64 encode
- cols: Maximum column width for line wrapping
"""
def write_entry(self, dn: str, entry: Union[Dict[str, Any], LDAPEntry]) -> None:
"""
Write LDAP entry to LDIF file.
Parameters:
- dn: Distinguished name of the entry
- entry: Dictionary or LDAPEntry with attributes and values
"""
def write_entries(self, entries: Iterable[Union[LDAPEntry, Tuple[str, Dict]]]) -> None:
"""
Write multiple LDAP entries to LDIF file.
Parameters:
- entries: Iterable of LDAPEntry objects or (dn, attributes) tuples
"""
def write_change_record(
self,
dn: str,
changetype: str,
change_data: Dict[str, Any]
) -> None:
"""
Write LDIF change record to file.
Parameters:
- dn: Distinguished name
- changetype: Operation type ("add", "modify", "delete", "modrdn")
- change_data: Change-specific data dictionary
"""
def write_version(self, version: int = 1) -> None:
"""
Write LDIF version line.
Parameters:
- version: LDIF version number (default: 1)
"""
def write_comment(self, comment: str) -> None:
"""
Write comment line to LDIF file.
Parameters:
- comment: Comment text (without # prefix)
"""
def write_dn(self, dn: str) -> None:
"""
Write DN line to LDIF file.
Parameters:
- dn: Distinguished name
"""
def write_attribute(self, name: str, value: Union[str, bytes]) -> None:
"""
Write single attribute line to LDIF file.
Parameters:
- name: Attribute name
- value: Attribute value (string or bytes)
"""
def flush(self) -> None:
"""Flush output buffer to file."""
@property
def cols(self) -> int:
"""Maximum column width for line wrapping."""
@property
def base64_attrs(self) -> Optional[List[str]]:
"""List of attributes that are always base64 encoded."""LDIF Exceptions
Specialized exception for LDIF parsing and writing errors.
from bonsai.ldif import LDIFError
class LDIFError(LDAPError):
"""Exception raised during LDIF file reading or writing operations."""
@property
def code(self) -> int:
"""Error code (-300 for LDIF errors)."""Usage Examples
Reading LDIF Files
from bonsai.ldif import LDIFReader
import io
# Read LDIF from file
with open('directory_export.ldif', 'r', encoding='utf-8') as f:
reader = LDIFReader(f)
# Parse entries
for dn, attributes in reader.parse():
print(f"Entry: {dn}")
for attr_name, values in attributes.items():
print(f" {attr_name}: {values}")
print()
# Read LDIF from string
ldif_data = '''
version: 1
dn: cn=John Doe,ou=people,dc=example,dc=com
objectClass: person
objectClass: organizationalPerson
cn: John Doe
sn: Doe
givenName: John
mail: john.doe@example.com
description: Software Engineer
dn: cn=Jane Smith,ou=people,dc=example,dc=com
objectClass: person
objectClass: organizationalPerson
cn: Jane Smith
sn: Smith
givenName: Jane
mail: jane.smith@example.com
telephoneNumber: +1-555-0123
'''
reader = LDIFReader(io.StringIO(ldif_data))
print(f"LDIF Version: {reader.version}")
# Parse into LDAPEntry objects
for entry in reader.parse_entry_records():
print(f"Entry DN: {entry.dn}")
print(f"Common Name: {entry['cn'][0]}")
print(f"Object Classes: {', '.join(entry['objectClass'])}")
if 'mail' in entry:
print(f"Email: {entry['mail'][0]}")
print()Writing LDIF Files
from bonsai.ldif import LDIFWriter
from bonsai import LDAPEntry
import io
# Create LDIF writer
output = io.StringIO()
writer = LDIFWriter(output)
# Write version
writer.write_version(1)
# Write comment
writer.write_comment("Directory export created on 2024-01-15")
# Create and write entries
people = [
{
'dn': 'cn=Alice Johnson,ou=people,dc=example,dc=com',
'objectClass': ['person', 'organizationalPerson', 'inetOrgPerson'],
'cn': 'Alice Johnson',
'sn': 'Johnson',
'givenName': 'Alice',
'mail': 'alice.johnson@example.com',
'employeeNumber': '12345'
},
{
'dn': 'cn=Bob Wilson,ou=people,dc=example,dc=com',
'objectClass': ['person', 'organizationalPerson'],
'cn': 'Bob Wilson',
'sn': 'Wilson',
'givenName': 'Bob',
'telephoneNumber': '+1-555-0456'
}
]
for person in people:
dn = person.pop('dn')
writer.write_entry(dn, person)
# Get LDIF output
ldif_output = output.getvalue()
print(ldif_output)
# Write to file
with open('people_export.ldif', 'w', encoding='utf-8') as f:
writer = LDIFWriter(f)
writer.write_version(1)
# Write LDAPEntry objects directly
for person in people:
entry = LDAPEntry(person['dn'])
for attr, value in person.items():
if attr != 'dn':
entry[attr] = value
writer.write_entry(str(entry.dn), entry)Processing Change Records
from bonsai.ldif import LDIFReader, LDIFWriter
import io
# LDIF with change records
change_ldif = '''
version: 1
# Add new user
dn: cn=New User,ou=people,dc=example,dc=com
changetype: add
objectClass: person
objectClass: organizationalPerson
cn: New User
sn: User
givenName: New
# Modify existing user
dn: cn=John Doe,ou=people,dc=example,dc=com
changetype: modify
replace: mail
mail: john.doe.new@example.com
-
add: telephoneNumber
telephoneNumber: +1-555-9999
-
# Delete user
dn: cn=Old User,ou=people,dc=example,dc=com
changetype: delete
# Rename user
dn: cn=Jane Smith,ou=people,dc=example,dc=com
changetype: modrdn
newrdn: cn=Jane Smith-Brown
deleteoldrdn: 1
newsuperior: ou=married,ou=people,dc=example,dc=com
'''
# Parse change records
reader = LDIFReader(io.StringIO(change_ldif))
for change in reader.parse_change_records():
dn, changetype, change_data = change
print(f"Change: {changetype} on {dn}")
if changetype == 'add':
print(" Adding new entry with attributes:")
for attr, values in change_data.items():
print(f" {attr}: {values}")
elif changetype == 'modify':
print(" Modifications:")
for mod in change_data['modifications']:
op, attr, values = mod
print(f" {op} {attr}: {values}")
elif changetype == 'delete':
print(" Deleting entry")
elif changetype == 'modrdn':
print(f" New RDN: {change_data['newrdn']}")
print(f" Delete old RDN: {change_data['deleteoldrdn']}")
if 'newsuperior' in change_data:
print(f" New superior: {change_data['newsuperior']}")
print()Binary Data Handling
from bonsai.ldif import LDIFReader, LDIFWriter
import base64
import io
# LDIF with binary data (base64 encoded)
binary_ldif = '''
version: 1
dn: cn=user-with-photo,ou=people,dc=example,dc=com
objectClass: person
objectClass: organizationalPerson
cn: user-with-photo
sn: Photo
jpegPhoto:: /9j/4AAQSkZJRgABAQEAYABgAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNA...
userCertificate:: MIIDXTCCAkWgAwIBAgIJAKoK...
'''
# Read binary data
reader = LDIFReader(io.StringIO(binary_ldif))
for dn, attributes in reader.parse():
print(f"Entry: {dn}")
for attr, values in attributes.items():
if attr in ['jpegPhoto', 'userCertificate']:
# Binary attributes are returned as bytes
for value in values:
if isinstance(value, bytes):
print(f" {attr}: <binary data, {len(value)} bytes>")
else:
print(f" {attr}: {value}")
else:
print(f" {attr}: {values}")
# Write binary data
output = io.StringIO()
writer = LDIFWriter(output, base64_attrs=['jpegPhoto', 'userCertificate'])
# Binary data will be automatically base64 encoded
binary_data = b'\x89PNG\r\n\x1a\n...' # PNG image data
writer.write_entry('cn=test,dc=example,dc=com', {
'objectClass': ['person'],
'cn': 'test',
'sn': 'user',
'jpegPhoto': binary_data # Will be base64 encoded automatically
})
print(output.getvalue())Batch Import/Export Operations
from bonsai import LDAPClient, LDAPEntry
from bonsai.ldif import LDIFReader, LDIFWriter
def export_to_ldif(client, base_dn, output_file):
"""Export LDAP directory tree to LDIF file."""
with client.connect() as conn:
# Search entire subtree
results = conn.search(base_dn, 2) # SUBTREE scope
with open(output_file, 'w', encoding='utf-8') as f:
writer = LDIFWriter(f)
writer.write_version(1)
writer.write_comment(f"Export of {base_dn} subtree")
for entry in results:
writer.write_entry(str(entry.dn), entry)
print(f"Exported {len(results)} entries to {output_file}")
def import_from_ldif(client, input_file):
"""Import entries from LDIF file to LDAP directory."""
with open(input_file, 'r', encoding='utf-8') as f:
reader = LDIFReader(f)
with client.connect() as conn:
imported_count = 0
for entry in reader.parse_entry_records():
try:
conn.add(entry)
imported_count += 1
print(f"Imported: {entry.dn}")
except Exception as e:
print(f"Failed to import {entry.dn}: {e}")
print(f"Successfully imported {imported_count} entries")
# Usage
client = LDAPClient("ldap://localhost")
client.set_credentials("SIMPLE", user="cn=admin,dc=example,dc=com", password="secret")
# Export directory to LDIF
export_to_ldif(client, "ou=people,dc=example,dc=com", "people_backup.ldif")
# Import from LDIF
import_from_ldif(client, "people_restore.ldif")Custom URL Handlers
from bonsai.ldif import LDIFReader
import io
import requests
def http_handler(url):
"""Custom handler for HTTP URLs."""
response = requests.get(url)
response.raise_for_status()
return response.content
def ftp_handler(url):
"""Custom handler for FTP URLs."""
# Implementation for FTP URL loading
pass
# LDIF with external URL references
url_ldif = '''
version: 1
dn: cn=user-with-cert,ou=people,dc=example,dc=com
objectClass: person
cn: user-with-cert
sn: Cert
userCertificate:< http://example.com/certs/user.crt
jpegPhoto:< file:///path/to/photo.jpg
'''
reader = LDIFReader(io.StringIO(url_ldif), autoload=True)
# Register custom URL handlers
reader.set_resource_handler('http', http_handler)
reader.set_resource_handler('https', http_handler)
reader.set_resource_handler('ftp', ftp_handler)
# Parse will automatically load external resources
for dn, attributes in reader.parse():
print(f"Entry: {dn}")
for attr, values in attributes.items():
if attr in ['userCertificate', 'jpegPhoto']:
for value in values:
if isinstance(value, bytes):
print(f" {attr}: <loaded {len(value)} bytes from URL>")
else:
print(f" {attr}: {values}")Install with Tessl CLI
npx tessl i tessl/pypi-bonsai