from __future__ import annotations from typing import Any from pydantic import BaseModel, Field, model_validator DEFAULT_LDAP_ATTRIBUTE_MAP: dict[str, str] = { "source_key": "entryUUID", "source_revision": "modifyTimestamp", "display_name": "displayName", "given_name": "givenName", "family_name": "sn", "organization": "o", "role_title": "title", "email": "mail", "phone": "telephoneNumber", "street": "streetAddress", "postal_code": "postalCode", "locality": "l", "region": "st", "country": "c", "tags": "memberOf", } class AddressLdapConnectionRequest(BaseModel): url: str = Field(min_length=1, max_length=2000) credential_ref: str | None = Field(default=None, max_length=1000) bind_dn: str | None = Field(default=None, max_length=1000) start_tls: bool = True connect_timeout: int = Field(default=10, ge=1, le=30) receive_timeout: int = Field(default=30, ge=1, le=120) class AddressLdapDiscoveryResponse(BaseModel): base_dns: list[str] = Field(default_factory=list) class AddressLdapSourceCreateRequest(AddressLdapConnectionRequest): display_name: str = Field(min_length=1, max_length=255) base_dn: str = Field(min_length=1, max_length=2000) search_filter: str = Field(default="(&(objectClass=person)(mail=*))", min_length=1, max_length=2000) page_size: int = Field(default=500, ge=1, le=1000) max_entries: int = Field(default=10_000, ge=1, le=10_000) attribute_map: dict[str, str] = Field(default_factory=lambda: dict(DEFAULT_LDAP_ATTRIBUTE_MAP), max_length=40) @model_validator(mode="after") def validate_mapping(self) -> "AddressLdapSourceCreateRequest": if "source_key" not in self.attribute_map: raise ValueError("LDAP mappings require a stable source_key attribute.") if not any(key in self.attribute_map for key in ("display_name", "email", "given_name", "family_name", "organization")): raise ValueError("LDAP mappings require at least one contact identity attribute.") if any(not key.strip() or not value.strip() for key, value in self.attribute_map.items()): raise ValueError("LDAP mapping names and attributes cannot be blank.") return self class AddressLdapTestResponse(BaseModel): success: bool base_dn: str sampled_entries: int attributes: list[str] = Field(default_factory=list) diagnostic: dict[str, Any] = Field(default_factory=dict) __all__ = [ "AddressLdapConnectionRequest", "AddressLdapDiscoveryResponse", "AddressLdapSourceCreateRequest", "AddressLdapTestResponse", "DEFAULT_LDAP_ATTRIBUTE_MAP", ]