or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

active-directory.mdasync-frameworks.mdconnection-pooling.mdcore-ldap.mdindex.mdldif-support.mdutilities-errors.md
tile.json

tessl/pypi-bonsai

Python 3 module for accessing LDAP directory servers with async framework support and Active Directory integration.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/bonsai@1.5.x

To install, run

npx @tessl/cli install tessl/pypi-bonsai@1.5.0

index.mddocs/

Bonsai

A comprehensive Python LDAP library for accessing LDAP directory servers with asynchronous framework support and Windows Active Directory integration. Bonsai provides a pythonic interface with high-performance C extensions, dictionary-like LDAP entry handling, and extensive async framework support.

Package Information

  • Package Name: bonsai
  • Language: Python
  • Installation: pip install bonsai

Core Imports

import bonsai

Common imports for LDAP operations:

from bonsai import LDAPClient, LDAPConnection, LDAPEntry, LDAPDN, LDAPURL

For async operations:

from bonsai.asyncio import AIOLDAPConnection
from bonsai.gevent import GeventLDAPConnection
from bonsai.tornado import TornadoLDAPConnection  
from bonsai.trio import TrioLDAPConnection

Basic Usage

from bonsai import LDAPClient

# Create LDAP client
client = LDAPClient("ldap://localhost")
client.set_credentials("SIMPLE", user="cn=admin,dc=example,dc=com", password="secret")

# Connect and perform operations
with client.connect() as conn:
    # Search for entries
    results = conn.search("dc=example,dc=com", 2)
    
    # Access entry attributes
    for entry in results:
        print(f"DN: {entry.dn}")
        print(f"Attributes: {dict(entry)}")
        
    # Create new entry
    new_entry = LDAPEntry("cn=newuser,dc=example,dc=com")
    new_entry['objectClass'] = ['person', 'organizationalPerson']
    new_entry['cn'] = 'newuser'
    new_entry['sn'] = 'User'
    
    # Add to directory
    conn.add(new_entry)

Architecture

Bonsai's architecture is built around several key components:

  • LDAPClient: Configuration and connection factory for LDAP servers
  • LDAPConnection: Active connection object providing all LDAP operations
  • LDAPEntry: Dictionary-like interface for LDAP entries with change tracking
  • LDAPDN: Distinguished name parsing and manipulation
  • LDAPURL: LDAP URL parsing and construction

The library uses native LDAP libraries (libldap on Unix, WinLDAP on Windows) written in C for high performance, while providing a pythonic interface with automatic change tracking and comprehensive error handling.

Capabilities

Core LDAP Operations

Essential LDAP functionality including client configuration, connection management, entry operations, distinguished name handling, and URL parsing. These components provide the foundation for all LDAP interactions.

class LDAPClient:
    def __init__(self, url: str) -> None: ...
    def connect(self, is_async: bool = False) -> LDAPConnection: ...
    def set_credentials(self, mechanism: str, **kwargs) -> None: ...

class LDAPConnection:
    def search(self, base: str, scope: int, filter_exp: str = None, **kwargs) -> list[LDAPEntry]: ...
    def add(self, entry: LDAPEntry) -> None: ...
    def modify(self, entry: LDAPEntry) -> None: ...
    def delete(self, dn: str) -> None: ...

class LDAPEntry:
    def __init__(self, dn: str) -> None: ...
    def __getitem__(self, key: str) -> LDAPValueList: ...
    def __setitem__(self, key: str, value) -> None: ...

Core LDAP Operations

Asynchronous Framework Support

Native integration with multiple Python async frameworks including asyncio, gevent, tornado, and trio. Each framework provides specialized connection classes optimized for their respective event loops.

class AIOLDAPConnection(LDAPConnection):
    async def search(self, base: str, scope: int, **kwargs) -> list[LDAPEntry]: ...
    async def add(self, entry: LDAPEntry) -> None: ...

class GeventLDAPConnection(LDAPConnection):
    def search(self, base: str, scope: int, **kwargs) -> list[LDAPEntry]: ...

class TornadoLDAPConnection(LDAPConnection):
    def search(self, base: str, scope: int, **kwargs) -> Future[list[LDAPEntry]]: ...

Async Framework Support

Active Directory Integration

Comprehensive Windows Active Directory support including Security Identifiers (SID), Access Control Lists (ACL), Security Descriptors, and User Account Control flags.

class SID:
    def __init__(self, sid_string: str) -> None: ...
    def __str__(self) -> str: ...

class SecurityDescriptor:
    def __init__(self, descriptor: bytes) -> None: ...
    @property
    def owner(self) -> SID: ...
    @property  
    def dacl(self) -> ACL: ...

class UserAccountControl:
    def __init__(self, value: int) -> None: ...
    @property
    def account_disabled(self) -> bool: ...

Active Directory Support

Connection Pooling

Connection pool management for handling multiple concurrent LDAP connections with automatic lifecycle management and configurable pool sizes.

class ConnectionPool:
    def __init__(self, client: LDAPClient, minconn: int = 1, maxconn: int = 10) -> None: ...
    def get(self) -> LDAPConnection: ...
    def put(self, conn: LDAPConnection) -> None: ...

Connection Pooling

LDIF Format Support

Read and write LDAP Data Interchange Format (LDIF) files for data import/export and backup operations.

class LDIFReader:
    def __init__(self, input_file) -> None: ...
    def parse(self) -> Iterator[tuple[str, dict]]: ...

class LDIFWriter:
    def __init__(self, output_file) -> None: ...
    def write_entry(self, dn: str, entry: dict) -> None: ...

LDIF Support

Utilities and Error Handling

Utility functions for escaping LDAP filter expressions and attribute values, plus comprehensive exception hierarchy for robust error handling.

def escape_filter_exp(expression: str) -> str: ...
def escape_attribute_value(value: str) -> str: ...
def get_vendor_info() -> dict: ...

class LDAPError(Exception):
    def __init__(self, error_message: str, error_code: int = None) -> None: ...

class AuthenticationError(LDAPError): ...
class ConnectionError(LDAPError): ...

Utilities and Errors

Types

from enum import IntEnum

class LDAPSearchScope(IntEnum):
    BASE = 0
    ONELEVEL = 1  
    SUBTREE = 2

class LDAPModOp(IntEnum):
    ADD = 0
    DELETE = 1
    REPLACE = 2

class LDAPValueList(list):
    def __init__(self, items=None) -> None: ...
    def append(self, item) -> None: ...
    def extend(self, items) -> None: ...
    def insert(self, index: int, item) -> None: ...
    def remove(self, item) -> None: ...
    def clear(self) -> None: ...
    
    @property
    def added(self) -> set: ...
    
    @property
    def deleted(self) -> set: ...
    
    @property
    def status(self) -> int: ...

class LDAPSearchIter:
    def __init__(self, conn: "LDAPConnection", base: str, scope: int, **kwargs) -> None: ...
    def __iter__(self) -> "LDAPSearchIter": ...
    def __next__(self) -> "LDAPEntry": ...
    def acquire_next_page(self) -> None: ...
    
    @property
    def cookie(self) -> Optional[bytes]: ...
    
    @property
    def estimated_list_count(self) -> int: ...

class LDAPReference:
    def __init__(self, client: "LDAPClient", references: List[Union[str, "LDAPURL"]]) -> None: ...
    
    @property
    def client(self) -> "LDAPClient": ...
    
    @property
    def references(self) -> List["LDAPURL"]: ...