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(model_cls: type[_LDAPActorT], limit: int = 10, **filters: str | None) → list[_LDAPActorT]

Fetch all items that match the given filters and parse to given model.

Parameters:

  • model_cls – Pydantic model class

  • limit – How many items to return

  • **filters – LDAP compatible filters, will be joined in AND-condition

Returns:

List of instances of model_cls

close() → None

Close the connector’s underlying LDAP connection.

get_functional_account(*, mail: str = '*', objectGUID: str = '*', sAMAccountName: str = '*', **filters: str | None) → LDAPActor

Get a single LDAP functional account for the given filters.

Parameters:

  • mail – Email address of the functional account

  • objectGUID – Internal LDAP identifier

  • sAMAccountName – Account name

  • **filters – Filters for LDAP search

Raises:

MExError – If number of LDAP entries that match the filters is not 1

Returns:

Single LDAP functional account matching the filters

get_functional_accounts(*, mail: str = '*', objectGUID: str = '*', sAMAccountName: str = '*', limit: int = 10, **filters: str | None) → list[LDAPActor]

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

  • objectGUID – Internal LDAP identifier

  • sAMAccountName – Account name

  • limit – How many items to return

  • **filters – Additional filters

Returns:

List of LDAP functional accounts

get_person(*, employeeID: str = '*', given_name: str = '*', mail: str = '*', objectGUID: str = '*', sAMAccountName: str = '*', surname: str = '*', **filters: str | None) → LDAPPerson

Get a single LDAP person for the given filters.

Parameters:

  • employeeID – Employee ID, must be present

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

  • mail – Email address, defaults to non-null

  • objectGUID – Internal LDAP identifier

  • sAMAccountName – str = “*”, # noqa: N803

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

  • **filters – Filters for LDAP search

Raises:

MExError – If number of LDAP entries that match the filters is not 1

Returns:

Single LDAP person matching the filters

get_persons(*, employeeID: str = '*', given_name: str = '*', mail: str = '*', objectGUID: str = '*', sAMAccountName: str = '*', surname: str = '*', limit: int = 10, **filters: str | None) → 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.

Only consider LDAP entries of objectClass ‘user’, ObjectCategory ‘Person’. Additional required attributes are: sAMAccountName, employeeID.

Parameters:

  • employeeID – Employee identifier

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

  • mail – Email address, defaults to non-null

  • objectGUID – Internal LDAP identifier

  • sAMAccountName – Account name

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

  • limit – How many items to return

  • **filters – Additional filters

Returns:

List of LDAP persons

mex.common.ldap.extract module

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

Return mapping from dynamic Person attribute to corresponding 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:

  • attribute – The key to use for the resulting mapping

  • persons – Iterable of LDAP persons

  • primary_source – Primary source for LDAP

Returns:

Mapping from a stringified LDAPPerson[attribute] to corresponding MergedPersonIdentifiers

mex.common.ldap.extract.get_ldap_persons(displayName: str | None, limit: int = 10) → list[LDAPPerson]

Get all ldap persons matching the filters.

Parameters:

  • displayName – Display name of a person

  • limit – How many items to return

Returns:

List of LDAP persons

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

Return a mapping from a person’s e-mail 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 LDP persons

  • primary_source – Primary source for LDAP

Returns:

Mapping from LDAPPerson.mail to corresponding MergedPersonIdentifiers

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(*, sAMAccountName: str | None = None, objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)], mail: list[Email] = [])

Bases: BaseModel

Model class for generic LDAP accounts.

static get_ldap_fields() → tuple[str, ...]

Return the fields that should be fetched from LDAP.

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.LDAPPerson(*, sAMAccountName: str | None = None, objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)], mail: list[Email] = [], company: str | None = None, department: str | None = None, departmentNumber: str | None = None, displayName: str | None = None, employeeID: str, givenName: Annotated[list[str], MinLen(min_length=1)], ou: list[str] = [], sn: str)

Bases: LDAPActor

Model class for LDAP persons.

company: str | None
department: str | None
departmentNumber: str | None
displayName: str | None
employeeID: str
classmethod get_ldap_fields() → tuple[str, ...]

Return the fields that should be fetched from LDAP.

givenName: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])]
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, metadata=[MinLen(min_length=1)]), '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
class mex.common.ldap.models.LDAPUnit(*, sAMAccountName: str | None = None, objectGUID: Annotated[UUID, UuidVersion(uuid_version=4)], mail: list[Email] = [], parent_label: str | None = None)

Bases: LDAPActor

Model class for LDAP organizational units.

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)]), 'parent_label': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), '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.

parent_label: str | None

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_ldap_actor_to_mex_contact_point(ldap_actor: LDAPActor, primary_source: ExtractedPrimarySource) → ExtractedContactPoint

Transform a single LDAPActor (a functional account) to an ExtractedContactPoint.

Parameters:

  • ldap_actor – LDAP actor

  • primary_source – Primary source for LDAP

Returns:

Extracted contact point

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

Transform LDAP actors (e.g. functional accounts) to ExtractedContactPoints.

Parameters:

  • ldap_actors – LDAP actors

  • primary_source – Primary source for LDAP

Returns:

List of extracted contact points

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_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_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.