mex.common.ldap package

Submodules

mex.common.ldap.connector module

class mex.common.ldap.connector.LDAPConnector

Bases: BaseConnector

Connector class to handle credentials and querying of LDAP.

__init__() → None

Create a new LDAP connection.

_fetch(search_filter: str, limit: int) → list[dict[str, Any]]

Fetch all items that match the given filters.

Parameters:

  • search_filter – LDAP search query

  • limit – How many items to return

Returns:

List of raw ldap items

static _sanitize(value: str) → str

Sanitize a query value by removing invalid chars.

_setup_connection() → Connection

Set up a new LDAP connection.

close() → None

Close the connector’s underlying LDAP connection.

get_functional_account(*, mail: str = '*') → LDAPFunctionalAccount

Get a single LDAP functional account for the given mail address.

Parameters:

mail – Email address of the functional account

Raises:
Returns:

Single LDAP functional account

get_functional_accounts(*, mail: str = '*', limit: int = 10) → list[LDAPFunctionalAccount]

Get LDAP functional accounts that match provided filters.

Some projects/resources declare functional mailboxes as their contact.

Parameters:

  • mail – Email address of the functional account

  • limit – How many items to return

Returns:

List of LDAP functional accounts

get_person(*, display_name: str = '*', employee_id: str = '*', given_name: str = '*', mail: str = '*', object_guid: str = '*', sam_account_name: str = '*', surname: str = '*') → LDAPPerson

Get a single LDAP person for the given filters.

Parameters:

  • display_name – Display name of the person

  • employee_id – Employee identifier

  • given_name – Given name of a person, defaults to non-null

  • mail – Email address, defaults to non-null

  • object_guid – Internal LDAP identifier

  • sam_account_name – Account name

  • surname – Surname of a person, defaults to non-null

Raises:
Returns:

Single LDAP person matching the filters

get_persons(*, display_name: str = '*', employee_id: str = '*', given_name: str = '*', mail: str = '*', object_guid: str = '*', sam_account_name: str = '*', surname: str = '*', limit: int = 10) → list[LDAPPerson]

Get LDAP persons that match the provided filters.

An LDAP person’s objectGUIDs is stable across name changes, whereas name based person identifiers of the schema SurnameF are not stable.

Parameters:

  • display_name – Display name of the person

  • employee_id – Employee identifier

  • given_name – Given name of a person, defaults to non-null

  • mail – Email address, defaults to non-null

  • object_guid – Internal LDAP identifier

  • sam_account_name – Account name

  • surname – Surname of a person, defaults to non-null

  • limit – How many items to return

Returns:

List of LDAP persons

get_persons_or_functional_accounts(*, query: str = '*', limit: int = 10) → list[LDAPPerson | LDAPFunctionalAccount]

Get LDAP persons or functional accounts.

Parameters:

  • query – Display name of person or email of functional account

  • limit – How many items to return

Returns:

List of LDAP persons and/or functional accounts

reconnect() → None

Close current ldap connection and initiate a new one.

mex.common.ldap.extract module

mex.common.ldap.extract.get_merged_ids_by_employee_ids(persons: Iterable[LDAPPerson], primary_source: ExtractedPrimarySource) → dict[str, list[MergedPersonIdentifier]]

Return a mapping from a person’s employeeID to their merged person ids.

MergedPersonIdentifiers are looked up in the identity provider and will be omitted for any person that has not yet been assigned an Identity there.

Parameters:

  • persons – Iterable of LDAP persons

  • primary_source – Primary source for LDAP

Returns:

Mapping from LDAPPerson.employeeID to corresponding MergedPersonIdentifiers

mex.common.ldap.extract.get_merged_ids_by_query_string(persons_with_query: Iterable[LDAPPersonWithQuery], primary_source: ExtractedPrimarySource) → dict[str, list[MergedPersonIdentifier]]

Return a mapping from a person query string to their merged person ids.

MergedPersonIdentifiers are looked up in the identity provider and will be omitted for any person that has not yet been assigned an Identity there.

Parameters:

  • persons_with_query – Iterable of LDP persons with query

  • primary_source – Primary source for LDAP

Returns:

Mapping from LDAPPersonWithQuery.query to corresponding MergedPersonIdentifiers

mex.common.ldap.models module

class mex.common.ldap.models.LDAPActor(*, objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)], sAMAccountName: str | None = None, mail: list[Email] = [])

Bases: BaseModel

Model class for generic LDAP accounts.

mail: list[Email]
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_max_length': 100000, 'str_min_length': 1, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'mail': FieldInfo(annotation=list[Email], required=False, default=[]), 'objectGUID': FieldInfo(annotation=UUID, required=True, metadata=[UuidVersion(uuid_version=4)]), 'sAMAccountName': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)]
sAMAccountName: str | None
class mex.common.ldap.models.LDAPFunctionalAccount(*, objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)], sAMAccountName: str | None = None, mail: list[Email] = [], ou: list[Literal['Funktion']])

Bases: LDAPActor

Model class for LDAP functional accounts.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_max_length': 100000, 'str_min_length': 1, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'mail': FieldInfo(annotation=list[Email], required=False, default=[]), 'objectGUID': FieldInfo(annotation=UUID, required=True, metadata=[UuidVersion(uuid_version=4)]), 'ou': FieldInfo(annotation=list[Literal['Funktion']], required=True), 'sAMAccountName': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

ou: list[Literal['Funktion']]
class mex.common.ldap.models.LDAPPerson(*, objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)], sAMAccountName: str | None = None, mail: list[Email] = [], employeeID: str, givenName: list[str], sn: str, company: str | None = None, department: str | None = None, departmentNumber: str | None = None, displayName: str | None = None, ou: list[str] = [])

Bases: LDAPActor

Model class for LDAP persons.

company: str | None
department: str | None
departmentNumber: str | None
displayName: str | None
employeeID: str
givenName: list[str]
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_max_length': 100000, 'str_min_length': 1, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'company': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'department': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'departmentNumber': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'displayName': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'employeeID': FieldInfo(annotation=str, required=True), 'givenName': FieldInfo(annotation=list[str], required=True), 'mail': FieldInfo(annotation=list[Email], required=False, default=[]), 'objectGUID': FieldInfo(annotation=UUID, required=True, metadata=[UuidVersion(uuid_version=4)]), 'ou': FieldInfo(annotation=list[str], required=False, default=[]), 'sAMAccountName': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'sn': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

ou: list[str]
sn: str
class mex.common.ldap.models.LDAPPersonWithQuery(*, person: LDAPPerson, query: str)

Bases: BaseModel

Wrapper bundling LDAPPerson models with the query string that found them.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_max_length': 100000, 'str_min_length': 1, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'person': FieldInfo(annotation=LDAPPerson, required=True), 'query': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

person: LDAPPerson
query: str

mex.common.ldap.transform module

class mex.common.ldap.transform.PersonName(surname: str = '*', given_name: str = '*', full_name: str = '')

Bases: object

Name of a person split into sur- and given-name.

full_name: str = ''
given_name: str = '*'
surname: str = '*'
mex.common.ldap.transform.analyse_person_string(string: str) → list[PersonName]

Try to extract a list of given- and surnames from a person string.

For supported formats of this implementation, check unittest.

Parameters:

string – Person string, containing their name in some form

Returns:

List of analyzed person names

mex.common.ldap.transform.transform_any_ldap_actor_to_extracted_persons_or_contact_points(ldap_actors: Iterable[LDAPPerson | LDAPFunctionalAccount], units: Iterable[ExtractedOrganizationalUnit], primary_source: ExtractedPrimarySource) → list[ExtractedPerson | ExtractedContactPoint]

Transform any LDAP actors to extracted person or contact points.

Parameters:

  • ldap_actors – Iterable of LDAP functional account or LDAP persons

  • units – Extracted organizational units

  • primary_source – Primary source for LDAP

Returns:

Extracted persons or contact points

mex.common.ldap.transform.transform_ldap_actor_to_mex_contact_point(ldap_functional_account: LDAPFunctionalAccount, primary_source: ExtractedPrimarySource) → ExtractedContactPoint

Transform a single LDAP functional account to an ExtractedContactPoint.

Parameters:

  • ldap_functional_account – LDAP functional account

  • primary_source – Primary source for LDAP

Returns:

Extracted contact point

mex.common.ldap.transform.transform_ldap_actors_to_mex_contact_points(ldap_functional_accounts: Iterable[LDAPFunctionalAccount], primary_source: ExtractedPrimarySource) → list[ExtractedContactPoint]

Transform LDAP functional accounts to ExtractedContactPoints.

Parameters:

  • ldap_functional_accounts – LDAP functional accounts

  • primary_source – Primary source for LDAP

Returns:

List of extracted contact points

mex.common.ldap.transform.transform_ldap_functional_account_to_extracted_contact_point(ldap_functional_account: LDAPFunctionalAccount, primary_source: ExtractedPrimarySource) → ExtractedContactPoint

Transform a single LDAP functional account to an ExtractedContactPoint.

Parameters:

  • ldap_functional_account – LDAP functional account

  • primary_source – Primary source for LDAP

Returns:

Extracted contact point

mex.common.ldap.transform.transform_ldap_functional_accounts_to_extracted_contact_points(ldap_functional_accounts: Iterable[LDAPFunctionalAccount], primary_source: ExtractedPrimarySource) → list[ExtractedContactPoint]

Transform LDAP functional accounts to ExtractedContactPoints.

Parameters:

  • ldap_functional_accounts – LDAP functional accounts

  • primary_source – Primary source for LDAP

Returns:

List of extracted contact points

mex.common.ldap.transform.transform_ldap_person_to_extracted_person(ldap_person: LDAPPerson, primary_source: ExtractedPrimarySource, units_by_identifier_in_primary_source: dict[str, ExtractedOrganizationalUnit]) → ExtractedPerson

Transform a single LDAP person to an ExtractedPerson.

Parameters:

  • ldap_person – LDAP person

  • primary_source – Primary source for LDAP

  • units_by_identifier_in_primary_source – Mapping to get units by LDAP ID

Returns:

Extracted person

mex.common.ldap.transform.transform_ldap_person_to_mex_person(ldap_person: LDAPPerson, primary_source: ExtractedPrimarySource, units_by_identifier_in_primary_source: dict[str, ExtractedOrganizationalUnit]) → ExtractedPerson

Transform a single LDAP person to an ExtractedPerson.

Parameters:

  • ldap_person – LDAP person

  • primary_source – Primary source for LDAP

  • units_by_identifier_in_primary_source – Mapping to get units by LDAP ID

Returns:

Extracted person

mex.common.ldap.transform.transform_ldap_persons_to_extracted_persons(ldap_persons: Iterable[LDAPPerson], primary_source: ExtractedPrimarySource, units: Iterable[ExtractedOrganizationalUnit]) → list[ExtractedPerson]

Transform LDAP persons to ExtractedPersons.

Parameters:

  • ldap_persons – LDAP persons

  • primary_source – Primary source for LDAP

  • units – Extracted organizational units

Returns:

List of extracted persons

mex.common.ldap.transform.transform_ldap_persons_to_mex_persons(ldap_persons: Iterable[LDAPPerson], primary_source: ExtractedPrimarySource, units: Iterable[ExtractedOrganizationalUnit]) → list[ExtractedPerson]

Transform LDAP persons to ExtractedPersons.

Parameters:

  • ldap_persons – LDAP persons

  • primary_source – Primary source for LDAP

  • units – Extracted organizational units

Returns:

List of extracted persons

mex.common.ldap.transform.transform_ldap_persons_with_query_to_extracted_persons(ldap_persons_with_query: Iterable[LDAPPersonWithQuery], primary_source: ExtractedPrimarySource, units: Iterable[ExtractedOrganizationalUnit]) → list[ExtractedPerson]

Transform LDAP persons with query to ExtractedPersons.

Parameters:

  • ldap_persons_with_query – LDAP persons with query

  • primary_source – Primary source for LDAP

  • units – Extracted organizational units

Returns:

List of extracted persons

mex.common.ldap.transform.transform_ldap_persons_with_query_to_mex_persons(ldap_persons_with_query: Iterable[LDAPPersonWithQuery], primary_source: ExtractedPrimarySource, units: Iterable[ExtractedOrganizationalUnit]) → list[ExtractedPerson]

Transform LDAP persons with query to ExtractedPersons.

Parameters:

  • ldap_persons_with_query – LDAP persons with query

  • primary_source – Primary source for LDAP

  • units – Extracted organizational units

Returns:

List of extracted persons

Module contents

Helper extractor to extract data from Lightweight Directory Access Protocol (LDAP).

Common use cases: - extract employee accounts of your organization - extract functional accounts of your organization

Possible queries are for example the account name, surname, given name, or email.

Configuration

For configuring the ldap connection, set the settings parameter ldap_url (see mex.common.settings for further info) to an LDAP url (see

Extracting data

Use the LDAPConnector from the ldap.connector module to extract data.

Transforming data

The module ldap.transform contains functions for transforming LDAP data into MEx models.

The mex_person.stableTargetId attribute can be used in any entity that requires a MergedPersonIdentifier.

Convenience Functions

The module ldap.extract holds convenience functions, e.g. for build a mapping from query strings to `stableTargetId`s.