mex.extractors.seq_repo package

Submodules

mex.extractors.seq_repo.extract module

mex.extractors.seq_repo.extract.extract_source_project_coordinator_by_name(seq_repo_sources: list[SeqRepoSource]) → dict[str, ExtractedPerson]

Extract Persons by their query string for source project coordinators.

Parameters:

seq_repo_sources – Seq Repo sources

Returns:

dictionary of Extracted persons by query

mex.extractors.seq_repo.extract.extract_sources() → list[SeqRepoSource]

Extract Seq Repo sources by loading data from source json file.

Returns:

List of Seq Repo resources

mex.extractors.seq_repo.main module

mex.extractors.seq_repo.model module

class mex.extractors.seq_repo.model.SeqRepoSource(*, project_coordinators: list[str], basepair_count: int, customer_org_unit_id: str | None = None, sequencing_date: str | None = None, igs_id: str | None = None, lims_sample_id: str, sequencing_platform: str | None = None, species: str | None = None, pathogen_code: str | None = None, project_name: str, customer_sample_name: str, project_id: str, reads_count: int, system_feedback: str | None = None)

Bases: BaseRawData

Model class for Seq Repo Source.

basepair_count: int
customer_org_unit_id: str | None
customer_sample_name: str
get_end_year() → TemporalEntity | None

Return end year from extractor.

get_identifier_in_primary_source() → str | None

Return identifier in primary source from extractor.

get_partners() → Sequence[__annotationlib_name_1__]

Return partners from extractor.

get_start_year() → TemporalEntity | None

Return start year from extractor.

get_units() → Sequence[__annotationlib_name_1__]

Return units from extractor.

igs_id: str | None
lims_sample_id: str
model_config = {'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].

pathogen_code: str | None
project_coordinators: list[str]
project_id: str
project_name: str
reads_count: int
sequencing_date: str | None
sequencing_platform: str | None
species: str | None
system_feedback: str | None

mex.extractors.seq_repo.settings module

class mex.extractors.seq_repo.settings.SeqRepoSettings(*, mapping_path: AssetsPath = AssetsPath('mappings/seq-repo'))

Bases: BaseModel

Settings submodel for the SeqRepo extractor.

mapping_path: AssetsPath
model_config = {'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].

mex.extractors.seq_repo.transform module

mex.extractors.seq_repo.transform.get_resolved_project_coordinators_and_units(project_coordinators: list[str], seq_repo_extracted_persons_by_name: dict[str, ExtractedPerson]) → tuple[list[MergedPersonIdentifier], list[MergedOrganizationalUnitIdentifier]]

Get ldap resolved ids of project coordinators and units.

Parameters:

  • project_coordinators – Seq Repo raw project coordinator names

  • seq_repo_extracted_persons_by_name – Seq Repo Sources resolved project coordinators as Extracted Persons

Returns:

Resolved ids project coordinator and units

mex.extractors.seq_repo.transform.transform_seq_repo_access_platform_to_extracted_access_platform(seq_repo_access_platform: AccessPlatformMapping) → ExtractedAccessPlatform

Transform seq-repo access platform to ExtractedAccessPlatform.

Parameters:

seq_repo_access_platform – Seq Repo access platform mapping model

Returns:

ExtractedAccessPlatform

mex.extractors.seq_repo.transform.transform_seq_repo_activities_to_extracted_activities(seq_repo_sources: list[SeqRepoSource], seq_repo_activity: ActivityMapping, seq_repo_extracted_persons_by_name: dict[str, ExtractedPerson]) → list[ExtractedActivity]

Transform seq-repo activities to list of unique ExtractedActivity.

Parameters:

  • seq_repo_sources – Seq Repo extracted sources

  • seq_repo_activity – Seq Repo activity mapping models with default values

  • seq_repo_extracted_persons_by_name – Seq Repo sources resolved project coordinators ldap query results

Returns:

list of unique ExtractedActivity

mex.extractors.seq_repo.transform.transform_seq_repo_resource_to_extracted_resource(seq_repo_sources: list[SeqRepoSource], seq_repo_activities: dict[str, ExtractedActivity], mex_access_platform: ExtractedAccessPlatform, seq_repo_resource: ResourceMapping, seq_repo_extracted_persons_by_name: dict[str, ExtractedPerson], extracted_organization_rki: ExtractedOrganization) → list[ExtractedResource]

Transform seq-repo resources to ExtractedResource.

Parameters:

  • seq_repo_sources – Seq Repo extracted sources

  • seq_repo_activities – Seq Repo extracted activity for default values from mapping

  • mex_access_platform – Extracted access platform

  • seq_repo_resource – Seq Repo resource mapping model with default values

  • seq_repo_extracted_persons_by_name – Seq Repo Sources resolved project coordinators as Extracted Persons

  • extracted_organization_rki – wikidata extracted organization

Returns:

list of ExtractedResource

Module contents