mex.extractors.blueant package¶

Subpackages¶

mex.extractors.blueant.models package

Submodules¶

mex.extractors.blueant.checks module¶

mex.extractors.blueant.connector module¶

class mex.extractors.blueant.connector.BlueAntConnector¶

Bases: HTTPConnector

Connector class to handle authentication and requesting the Blue Ant API.

_get_json_from_api(relative_url: str) → dict[str, Any]¶

Get json from blueant api.

Parameters:: relative_url (str) – relative url of the api
Raises:: MExError if Blue Ant API returns an error in the response body –
Returns:: Parsed JSON body of the response

_set_authentication() → None¶: Authenticate to the host.

_set_url() → None¶: Set url of the host.

get_client_name(client_id: int) → str¶

Get client name for client id.

Parameters:: client_id – int: id of the client
Returns:: name of the client
Return type:: str

get_department_name(department_id: int) → str¶

Get name for department id.

Parameters:: department_id – int: id of the department
Returns:: name of the department
Return type:: str

get_persons() → list[BlueAntPerson]¶: Get map of Blue Ant person IDs to employee IDs.

get_projects() → Generator[BlueAntProject, None, None]¶

Load Blue Ant sources by querying the Blue Ant API projects endpoint.

Returns:: Generator for Blue Ant projects

get_status_name(status_id: int) → str¶

Get name for status id.

Parameters:: status_id – int: id of the status
Returns:: name of the status
Return type:: str

get_type_description(type_id: int) → str¶

Get description for type id.

Parameters:: type_id – int: id of the type
Returns:: description of the type
Return type:: str

mex.extractors.blueant.extract module¶

mex.extractors.blueant.extract.extract_blueant_organizations(blueant_sources: list[BlueAntSource]) → dict[str, MergedOrganizationIdentifier]¶

Search and extract organization from wikidata.

Parameters:: blueant_sources – Iterable of blueant sources
Returns:: Dict with organization label and WikidataOrganization ID

mex.extractors.blueant.extract.extract_blueant_project_leaders(blueant_sources: Iterable[BlueAntSource]) → Generator[LDAPPerson, None, None]¶

Extract LDAP persons for Blue Ant project leaders.

Parameters:: blueant_sources – Blue Ant sources
Returns:: Generator for LDAP persons

mex.extractors.blueant.extract.extract_blueant_sources() → Generator[BlueAntSource, None, None]¶

Load Blue Ant sources from Blue Ant API.

Returns:: Generator for Blue Ant sources

mex.extractors.blueant.extract.remove_prefixes_from_name(name: str) → str¶

Remove prefix according to settings.

Parameters:: name – string containing project label

Settings:: blueant.delete_prefixes: delete prefixes of labels starting with these terms

Returns:: string cleaned of prefixes

mex.extractors.blueant.filter module¶

mex.extractors.blueant.filter.filter_and_log_blueant_source(source: BlueAntSource) → bool¶

Filter a BlueantSource according to settings and log filtering.

Parameters:: source – BlueantSource

Settings:: blueant.skip_labels: Skip source if these terms are in the label

Returns:: False if source is filtered out, else True

mex.extractors.blueant.filter.filter_and_log_blueant_sources(sources: Iterable[BlueAntSource]) → Generator[BlueAntSource, None, None]¶

Filter Blueant sources and log filtered sources.

Parameters:: sources – BlueantSources Generator
Returns:: Generator for Blue Ant sources

mex.extractors.blueant.main module¶

mex.extractors.blueant.settings module¶

class mex.extractors.blueant.settings.BlueAntSettings(*, api_key: SecretStr = SecretStr('**********'), url: str = 'https://blueant', skip_labels: list[str] = ['test'], delete_prefixes: list[str] = ['_', '1_', '2_', '3_', '4_', '5_', '6_', '7_', '8_', '9_'], mapping_path: AssetsPath = AssetsPath('mappings/blueant'))¶

Bases: BaseModel

Blue Ant settings submodel definition for the Blue Ant extractor.

api_key: SecretStr¶

delete_prefixes: list[str]¶

mapping_path: AssetsPath¶

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_by_alias': True, 'validate_by_name': True, 'validate_default': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

skip_labels: list[str]¶

url: str¶

mex.extractors.blueant.transform module¶

mex.extractors.blueant.transform.transform_blueant_sources_to_extracted_activities(blueant_sources: Iterable[BlueAntSource], person_stable_target_ids_by_employee_id: dict[str, list[MergedPersonIdentifier]], unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier], activity: ActivityMapping, blueant_merged_organization_ids_by_query_string: dict[str, MergedOrganizationIdentifier]) → Generator[ExtractedActivity, None, None]¶

Transform Blue Ant sources to ExtractedActivities.

Parameters:

blueant_sources – Blue Ant sources
person_stable_target_ids_by_employee_id – Mapping from LDAP employeeIDs to person stable target IDs
unit_stable_target_ids_by_synonym – Map from unit acronyms and labels to unit stable target IDs
activity – activity mapping model with default values
blueant_merged_organization_ids_by_query_string – extracted blueant organizations dict

Returns:

Generator for ExtractedActivity instances