mex.extractors.publisher package

Submodules

mex.extractors.publisher.extract module

mex.extractors.publisher.extract.get_publishable_merged_item(identifier: AnyMergedIdentifier) → AnyMergedModel

Fetch a merged item from backend identified by its identifier.

Parameters:

identifier – Identifier of merged item of any entity Type

Returns:

the merged item of the identitifer

Raises:

MExError if not exactly one item is found. –

mex.extractors.publisher.extract.get_publishable_merged_items(*, query_string: str | None = None, entity_type: list[str] | None = None, reference_filters: list[ReferenceFilter] | None = None) → list[AnyMergedModel]

Read publishable merged items from backend.

mex.extractors.publisher.fields module

mex.extractors.publisher.filter module

mex.extractors.publisher.filter.cluster_and_filter_bibliographic_resources_by_unit(merged_bibliographic_resources: list[MergedBibliographicResource]) → dict[MergedOrganizationalUnitIdentifier, list[MergedBibliographicResource]]

Sort Bibliographic Resources by unit and filter out ‘forbidden’ units.

This function gets all the unpublishable units from the __all__/activity_filter. Then it collects all units, which are departments (i.e. direct child units of PRAES) Then it sorts all Bibliographic Resources into a dict by department if the department or its child units contributed and if that contributing unit is not an unpublishable unit.

Parameters:

merged_bibliographic_resources – Merged Bibliographic Resources as list

Returns:

dictionary of Bibliographic Resources by allowed units

Filter for persons with approving consent. Fail if a person has > 1 consent.

Parameters:

  • person_items – list of persons

  • consent_items – list of consents

Raises:

MExError if any person has more than one consent. The error lists all concerned – persons and consents.

Returns:

list of filtered persons with approving consent.

mex.extractors.publisher.main module

mex.extractors.publisher.models module

class mex.extractors.publisher.models.BibliographicResourceForCsv(*, contributingUnit: list[str] | None = None, publicationYear: str | None = None, creator: list[str], title: list[str], journal: list[str] | None = None, doi: str | None = None, accessRestriction: str, publisher: list[str] | None = None)

Bases: BaseModel

pydantic model of Bibliographic Resources for writing to csv.

accessRestriction: str
contributingUnit: list[str] | None
creator: list[str]
doi: str | None
journal: list[str] | None
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].

publicationYear: str | None
publisher: list[str] | None
title: list[str]
class mex.extractors.publisher.models.PublisherItemsLike(*args, **kwargs)

Bases: Protocol

Structural PublisherItem type to conform to Dagster asset boundaries.

items: list[MergedAccessPlatform | MergedActivity | MergedBibliographicResource | MergedConsent | MergedContactPoint | MergedDistribution | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedResourceSeries | MergedVariable | MergedVariableGroup]

mex.extractors.publisher.settings module

class mex.extractors.publisher.settings.PublisherSettings(*, skip_entity_types: list[str] = ['MergedPrimarySource', 'MergedConsent'], allowed_person_primary_sources: list[str] = ['endnote'], mapping_path: AssetsPath = AssetsPath('mappings'), departments_for_publications_csv: list[str] = ['PRNT', 'fg99'], sink: str = 's3')

Bases: BaseModel

Settings submodel definition for the publishing pipeline.

allowed_person_primary_sources: list[str]
departments_for_publications_csv: list[str]
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].

sink: str
skip_entity_types: list[str]

mex.extractors.publisher.transform module

mex.extractors.publisher.transform.get_resolved_names(identifier: MergedAccessPlatformIdentifier | MergedActivityIdentifier | MergedBibliographicResourceIdentifier | MergedConsentIdentifier | MergedContactPointIdentifier | MergedDistributionIdentifier | MergedOrganizationalUnitIdentifier | MergedOrganizationIdentifier | MergedPersonIdentifier | MergedPrimarySourceIdentifier | MergedResourceIdentifier | MergedResourceSeriesIdentifier | MergedVariableGroupIdentifier | MergedVariableIdentifier, field: str) → str

Get names from referenced merged item ids.

Parameters:

  • identifier – identifier of referenced merged item of which to get the name.

  • entitiy_type – entity type of the merged item.

  • field – field of the merged item in which the name needs to be looked up.

Returns:

name as string or None if not found.

Raises:

MExError if name or merged item have the wrong type. –

mex.extractors.publisher.transform.get_unit_id_per_person(publisher_merged_ldap_persons: list[MergedPerson], publisher_contact_points_and_units: PublisherItemsLike) → dict[MergedPersonIdentifier, list[MergedOrganizationalUnitIdentifier]]

For each Person get their unit IDs if the unit has an email address.

Parameters:

  • publisher_merged_ldap_persons – Merged Persons with primary source ldap

  • publisher_contact_points_and_units – Items container of units + contact points

Returns:

dictionary of unit identifiers by person identifier

mex.extractors.publisher.transform.transform_merged_bibliographic_resources_for_csv(merged_bibliographic_resources_by_unit: dict[MergedOrganizationalUnitIdentifier, list[MergedBibliographicResource]]) → dict[str, list[BibliographicResourceForCsv]]

Transform merged bibliographic resources to bibliographic resources for csv.

Parameters:

merged_bibliographic_resources_by_unit – dictionary of merged bibliographic resources by department unit ids.

Returns:

dictionary of BibliographicResourceForCsv entries by department unit short names

mex.extractors.publisher.transform.update_actor_references_where_needed(item: AnyMergedModel, allowed_actors: Collection[__annotationlib_name_1__], fallback_contact_identifiers: list[MergedContactPointIdentifier], fallback_unit_identifiers_by_person: dict[MergedPersonIdentifier, list[MergedOrganizationalUnitIdentifier]]) → None

Update references to actors, where needed.

We filter all fields that allow Person references to only contain references to publishable actors. For fields that also allow organizational units, non-consenting persons can get replaced by their organizational unit if the unit provides an email address. Fields that allow contact points, but contain no valid references are set to a fallback contact point. Should the field be required, not allow contact points, but still contain no valid references, we keep the broken ones in order to keep mex-model compliance. Would we skip those items instead, we might break other items relying on the former item, and start a recursive de-publication process - which we don’t want.

Module contents