ldapweb/webapp/pools.py
2024-07-07 21:46:29 +02:00

28 lines
762 B
Python

import ldap
from webapp import lru, config
from typing import Optional, Any
Pool = lru.LRUPool[str, ldap.ldapobject]
class LDAPConnectionPool(Pool):
def __init__(self, url: str, use_tls: bool = True, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.use_tls = use_tls
self.url = url
def bind(self, dn: str, password: str) -> ldap.ldapobject:
self.lock.acquire()
try:
conn = ldap.initialize(self.url)
if self.use_tls:
conn.start_tls_s()
conn.simple_bind_s(dn, password)
return self._insert(dn, conn)
finally:
self.lock.release()
def unbind(self, dn: str) -> Optional[ldap.ldapobject]:
return self.drop(dn)