mex.extractors.grippeweb package

Submodules

mex.extractors.grippeweb.connector module

class mex.extractors.grippeweb.connector.GrippewebConnector

Bases: BaseConnector

Connector to handle authentication and queries for Grippeweb SQL server.

__init__() → None

Create a new connector instance.

close() → None

Close the underlying connection.

parse_columns_by_column_name(table_name: str) → dict[str, list[Any]]

Execute whitelisted queries and zip results to column name.

mex.extractors.grippeweb.extract module

mex.extractors.grippeweb.extract.extract_columns_by_table_and_column_name() → dict[str, dict[str, list[Any]]]

Extract sql tables and parse them into column lists by table and column name.

Returns:

list of columns by column names and table names

mex.extractors.grippeweb.extract.extract_grippeweb_organizations(grippeweb_resource_mappings: list[ResourceMapping]) → dict[str, MergedOrganizationIdentifier]

Search and extract grippeweb organization from wikidata.

Parameters:

grippeweb_resource_mappings – grippeweb resource mapping models

Returns:

mapping default values

and values: MergedOrganizationIdentifier

Return type:

Dict with keys

mex.extractors.grippeweb.extract.extract_ldap_actors_for_functional_accounts(grippeweb_resource_mappings: list[ResourceMapping]) → list[LDAPFunctionalAccount]

Extract LDAP actors functional accounts from grippeweb resource mapping contacts.

Parameters:

grippeweb_resource_mappings – list of resources default value mapping models

Returns:

list of LDAP actors

mex.extractors.grippeweb.extract.extract_ldap_persons(grippeweb_resource_mappings: list[ResourceMapping], grippeweb_access_platform: AccessPlatformMapping) → list[LDAPPerson]

Extract LDAP persons for grippeweb.

Parameters:

  • grippeweb_resource_mappings – list of resources default value mapping models

  • grippeweb_access_platform – grippeweb access platform mapping model

Returns:

list of LDAP persons

mex.extractors.grippeweb.main module

mex.extractors.grippeweb.settings module

class mex.extractors.grippeweb.settings.GrippewebSettings(*, mapping_path: AssetsPath = AssetsPath('mappings/grippeweb'), mssql_connection_dsn: str = 'DRIVER={ODBC Driver 18 for SQL Server};SERVER=domain.tld;DATABASE=database')

Bases: BaseModel

Settings definition for the infection protection act data.

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

mssql_connection_dsn: str

mex.extractors.grippeweb.transform module

mex.extractors.grippeweb.transform.get_or_create_external_partner(resource: ResourceMapping, grippeweb_merged_organization_ids_by_query_str: dict[str, MergedOrganizationIdentifier]) → list[MergedOrganizationIdentifier] | None

Get external partner from wikidata or create organization.

Parameters:

  • resource – grippeweb resource mapping model

  • grippeweb_merged_organization_ids_by_query_str – extracted grippeweb organizations dict

Returns:

dict extracted grippeweb resource by identifier in primary source

mex.extractors.grippeweb.transform.transform_grippeweb_access_platform_to_extracted_access_platform(grippeweb_access_platform: AccessPlatformMapping, unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier], grippeweb_extracted_persons: list[ExtractedPerson]) → ExtractedAccessPlatform

Transform grippeweb access platform to ExtractedAccessPlatform.

Parameters:

  • grippeweb_access_platform – grippeweb access platform mapping model

  • unit_stable_target_ids_by_synonym – Unit stable target ids by synonym

  • grippeweb_extracted_persons – extracted grippeweb persons

Returns:

ExtractedAccessPlatform grippeweb

mex.extractors.grippeweb.transform.transform_grippeweb_resource_mappings_to_dict(grippeweb_resource_mappings: list[ResourceMapping], unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier], grippeweb_extracted_access_platform: ExtractedAccessPlatform, grippeweb_extracted_persons: list[ExtractedPerson], grippeweb_merged_organization_ids_by_query_str: dict[str, MergedOrganizationIdentifier], grippeweb_merged_contact_point_id_by_email: dict[str, MergedContactPointIdentifier]) → tuple[ExtractedResource, ExtractedResource]

Transform grippe web values to extracted resources.

Parameters:

  • grippeweb_resource_mappings – grippeweb resource mapping models

  • unit_stable_target_ids_by_synonym – merged organizational units by name

  • grippeweb_extracted_access_platform – extracted grippeweb access platform

  • grippeweb_extracted_persons – extracted grippeweb mex persons

  • grippeweb_merged_organization_ids_by_query_str – extracted grippeweb organizations dict

  • grippeweb_merged_contact_point_id_by_email – extracted grippeweb mex functional accounts

Returns:

tuple of grippeweb resources

mex.extractors.grippeweb.transform.transform_grippeweb_resource_mappings_to_extracted_resources(grippeweb_resource_mappings: list[ResourceMapping], unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier], grippeweb_extracted_access_platform: ExtractedAccessPlatform, grippeweb_extracted_persons: list[ExtractedPerson], grippeweb_merged_organization_ids_by_query_str: dict[str, MergedOrganizationIdentifier], grippeweb_merged_contact_point_id_by_email: dict[str, MergedContactPointIdentifier]) → tuple[ExtractedResource, ExtractedResource]

Transform grippe web values to extracted resources and link them.

Parameters:

  • grippeweb_resource_mappings – grippeweb resource mapping models

  • unit_stable_target_ids_by_synonym – merged organizational units by name

  • grippeweb_extracted_access_platform – extracted grippeweb access platform

  • grippeweb_extracted_persons – extracted grippeweb mex persons

  • grippeweb_merged_organization_ids_by_query_str – extracted grippeweb organizations dict

  • grippeweb_merged_contact_point_id_by_email – extracted grippeweb mex functional accounts

Returns:

tuple of grippeweb resources

mex.extractors.grippeweb.transform.transform_grippeweb_variable_group_to_extracted_variable_groups(grippeweb_variable_group: VariableGroupMapping, grippeweb_columns: dict[str, dict[str, list[Any]]], grippeweb_extracted_parent_resource: ExtractedResource) → list[ExtractedVariableGroup]

Transform Grippeweb variable groups to extracted variable groups.

Parameters:

  • grippeweb_variable_group – grippeweb variable group mapping model

  • grippeweb_columns – grippeweb data by column and table

  • grippeweb_extracted_parent_resource – extracted parent resource

Returns:

list of extracted variable groups

mex.extractors.grippeweb.transform.transform_grippeweb_variable_to_extracted_variables(grippeweb_variable: VariableMapping, grippeweb_extracted_variables_groups: list[ExtractedVariableGroup], grippeweb_columns: dict[str, dict[str, list[Any]]], grippeweb_extracted_parent_resource: ExtractedResource) → list[ExtractedVariable]

Transform Grippeweb variables to extracted variables.

Parameters:

  • grippeweb_variable – grippeweb variable mapping model

  • grippeweb_extracted_variables_groups – extracted grippeweb variable groups

  • grippeweb_columns – grippeweb data by column and table

  • grippeweb_extracted_parent_resource – extracted parent resource

Returns:

list of extracted variables

Module contents