mex.ifsg package

Submodules

mex.ifsg.connector module

class mex.ifsg.connector.IFSGConnector

Bases: BaseConnector

Connector to handle authentication and queries towards the IFSG SQL server.

__init__() → None

Create a new connector instance.

close() → None

Close the underlying connection.

parse_rows(model: type[BaseModel]) → list[dict[str, Any]]

Execute whitelisted queries and zip results to column name.

class mex.ifsg.connector.NoOpPyodbc

Bases: object

No-op pyodbc drop-in for when the libodbc dependency is not installed.

connect(_: str) → None

Create a new ODBC connection to a database.

mex.ifsg.extract module

mex.ifsg.extract.extract_sql_table(model: type[ModelT]) → list[ModelT]

Extract sql tables and parse them into pydantic models.

Returns:

list of parsed pydantic ModelT

mex.ifsg.filter module

mex.ifsg.filter.filter_empty_statement_area_group(meta_field: list[MetaField]) → list[MetaField]

Filter out meta_field rows with emtpy statement area group.

Parameters:

meta_field – list of MetaField entities

Returns:

filtered list of MetaField entities for variableGroup transformation

mex.ifsg.filter.filter_id_type_of_diseases(id_types_with_max_id_schema: list[int], meta_type: list[MetaType]) → list[int]

Filter id_types that correspond to a disease.

Parameters:

  • id_types_with_max_id_schema – id_types with latest id_schema

  • meta_type – MetaType list

Returns:

id_type list related to diseases

mex.ifsg.filter.filter_id_type_with_max_id_schema(meta_schema2type: list[MetaSchema2Type], max_id_schema: int) → list[int]

Filter for id_types with max id_schema.

Parameters:

  • meta_schema2type – MetaSchema2Type list

  • max_id_schema – latest id_schema

Returns:

id_type list filtered for latest id_schema

mex.ifsg.filter.filter_variables(meta_field: list[MetaField], meta_schema2field: list[MetaSchema2Field], max_id_schema: int, id_types_of_diseases: list[int]) → list[MetaField]

Filter meta_field list for variable transformation.

Parameters:

  • meta_field – list of MetaField entities

  • meta_schema2field – MetaSchema2Field list

  • max_id_schema – lates id_schema,

  • id_types_of_diseases – id_types related to diseases

Returns:

filtered list of MetaField entities to transform into variables

mex.ifsg.filter.get_max_id_schema(meta_schema2type: list[MetaSchema2Type]) → int

Return the latest id_schema.

Parameters:

meta_schema2type – MetaSchema2Type list

Returns:

latest id_schema

mex.ifsg.main module

mex.ifsg.settings module

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

Bases: BaseModel

Settings submodel definition for the infection protection act data.

mapping_path: AssetsPath
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]] = {'mapping_path': FieldInfo(annotation=AssetsPath, required=False, default=AssetsPath("mappings/__final__/ifsg"), description='Path to the directory with the ifsg mapping files containing the default values, absolute path or relative to `assets_dir`.'), 'mssql_connection_dsn': FieldInfo(annotation=str, required=False, default='DRIVER={ODBC Driver 18 for SQL Server};SERVER=domain.tld;DATABASE=database', description='Connection string for the ODBC Driver for SQL Server: https://learn.microsoft.com/en-us/sql/connect/odbc/dsn-connection-string-attribute')}

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

This replaces Model.__fields__ from Pydantic V1.

mssql_connection_dsn: str

mex.ifsg.transform module

mex.ifsg.transform.get_instrument_tool_or_apparatus(meta_disease: MetaDisease, resource_disease: dict[str, Any]) → list[Text]

Calculate instrument_tool_or_apparatus for MetaDisease reference definitions.

Parameters:

  • meta_disease – MetaDisease

  • resource_disease – resource_disease default values

Returns:

instrument_tool_or_apparatus list

mex.ifsg.transform.transform_ifsg_data_to_mex_variable_group(ifsg_variable_group: dict[str, Any], extracted_ifsg_resource_disease: list[ExtractedResource], extracted_primary_source: ExtractedPrimarySource, meta_field: list[MetaField], id_types_of_diseases: list[int]) → list[ExtractedVariableGroup]

Transform ifsg data to mex VariableGroup.

Parameters:

  • ifsg_variable_group – ifsg_variable_group default values

  • extracted_ifsg_resource_disease – ExtractedResource disease list

  • extracted_primary_source – ExtractedPrimarySource

  • meta_field – MetaField list

  • id_types_of_diseases – disease related id_types

Returns:

transform resource parent to ExtractedResource

mex.ifsg.transform.transform_ifsg_data_to_mex_variables(filtered_variables: list[MetaField], extracted_ifsg_resource_disease: list[ExtractedResource], extracted_ifsg_variable_group: list[ExtractedVariableGroup], extracted_primary_sources_ifsg: ExtractedPrimarySource, meta_catalogue2item: list[MetaCatalogue2Item], meta_catalogue2item2schema: list[MetaCatalogue2Item2Schema], meta_item: list[MetaItem]) → list[ExtractedVariable]

Transform ifsg data to mex Variable.

Parameters:

  • filtered_variables – MetaField list to transform into variables

  • extracted_ifsg_resource_disease – ExtractedResource disease list

  • extracted_ifsg_variable_group – variable group default values

  • extracted_primary_sources_ifsg – ExtractedPrimarySource

  • meta_catalogue2item – MetaCatalogue2Item list

  • meta_catalogue2item2schema – MetaCatalogue2Item2Schema list

  • meta_item – MetaItem list

Returns:

transform filtered variable to extracted variables

mex.ifsg.transform.transform_resource_disease_to_mex_resource(resource_disease: dict[str, Any], extracted_ifsg_resource_parent: ExtractedResource, extracted_ifsg_resource_state: list[ExtractedResource], meta_disease: list[MetaDisease], meta_type: list[MetaType], id_type_of_diseases: list[int], extracted_primary_source: ExtractedPrimarySource, unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier], extracted_organization_rki: ExtractedOrganization) → list[ExtractedResource]

Transform resource disease to mex resource.

Parameters:

  • resource_disease – resource_disease default values

  • extracted_ifsg_resource_parent – ExtractedResource

  • extracted_ifsg_resource_state – ExtractedResource

  • meta_disease – MetaDisease

  • meta_type – MetaType

  • id_type_of_diseases – list of disease related id_types

  • extracted_primary_source – ExtractedPrimarySource

  • unit_stable_target_ids_by_synonym – mapping unit synonyms to MergedOrganizationalUnitIdentifier

  • extracted_organization_rki – extracted organization for RKI

Returns:

transform resource disease to ExtractedResource list

mex.ifsg.transform.transform_resource_disease_to_mex_resource_row(id_type: int, resource_disease: dict[str, Any], extracted_ifsg_resource_parent: ExtractedResource, extracted_primary_source: ExtractedPrimarySource, stable_target_id_by_bundesland_id: dict[str, MergedResourceIdentifier], meta_disease_row_by_id_type: dict[int, MetaDisease], bundesland_by_in_bundesland: dict[str, Text], code_by_id_type: dict[int, str], unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier], extracted_organization_rki: ExtractedOrganization) → ExtractedResource

Transform resource disease row to mex resource.

Parameters:

  • id_type – id_type of Resource row

  • resource_disease – resource_disease default values

  • extracted_ifsg_resource_parent – ExtractedResource

  • extracted_primary_source – ExtractedPrimarySource

  • stable_target_id_by_bundesland_id – stable target id ti bundesland_id map

  • meta_disease_row_by_id_type – id type to meta disease row map

  • bundesland_by_in_bundesland – in bundesland str to bundesland Text map

  • code_by_id_type – id type to code map

  • unit_stable_target_ids_by_synonym – mapping unit synonyms to MergedOrganizationalUnitIdentifier

  • extracted_organization_rki – extracted organization for RKI

Returns:

transform resource disease row to ExtractedResource

mex.ifsg.transform.transform_resource_parent_to_mex_resource(resource_parent: dict[str, Any], extracted_primary_source: ExtractedPrimarySource, unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier]) → ExtractedResource

Transform resource parent to mex resource.

Parameters:

  • resource_parent – resource_parent default values

  • extracted_primary_source – ExtractedPrimarySource

  • unit_stable_target_ids_by_synonym – mapping unit synonyms to MergedOrganizationalUnitIdentifier

Returns:

resource parent transformed to ExtractedResource

mex.ifsg.transform.transform_resource_state_to_mex_resource(resource_state: dict[str, Any], extracted_ifsg_resource_parent: ExtractedResource, extracted_primary_source: ExtractedPrimarySource, unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier]) → list[ExtractedResource]

Transform resource state to mex resource.

Parameters:

  • resource_state – resource_state default values

  • extracted_ifsg_resource_parent – ExtractedResource

  • extracted_primary_source – ExtractedPrimarySource

  • unit_stable_target_ids_by_synonym – mapping unit synonyms to MergedOrganizationalUnitIdentifier

Returns:

transform resource state to ExtractedResource list

Module contents