mex.common.identity package

Submodules

mex.common.identity.base module

class mex.common.identity.base.BaseProvider

Bases: BaseConnector

Base class to define the interface of identity providers.

abstract assign(had_primary_source: MergedPrimarySourceIdentifier, identifier_in_primary_source: str) → Identity

Find an Identity in a database or assign a new one.

abstract fetch(*, had_primary_source: MergedPrimarySourceIdentifier | None = None, identifier_in_primary_source: str | None = None, stable_target_id: MergedAccessPlatformIdentifier | MergedActivityIdentifier | MergedBibliographicResourceIdentifier | MergedConsentIdentifier | MergedContactPointIdentifier | MergedDistributionIdentifier | MergedOrganizationalUnitIdentifier | MergedOrganizationIdentifier | MergedPersonIdentifier | MergedPrimarySourceIdentifier | MergedResourceIdentifier | MergedVariableGroupIdentifier | MergedVariableIdentifier | None = None) → list[Identity]

Find Identity instances matching the given filters.

mex.common.identity.memory module

class mex.common.identity.memory.MemoryIdentityProvider

Bases: BaseProvider

Connector class to handle read/write to the identity database.

__init__() → None

Initialize an in-memory database with the identity of MEx itself.

static _get_identifier(*args: str) → Identifier

Get deterministic identifier based on args.

assign(had_primary_source: MergedPrimarySourceIdentifier, identifier_in_primary_source: str) → Identity

Find an Identity in the in-memory database or assign a new one.

Parameters:

  • had_primary_source – Stable target ID of primary source

  • identifier_in_primary_source – Identifier in the primary source

Returns:

Newly created or updated Identity instance

close() → None

Trash the in-memory identity database.

fetch(*, had_primary_source: MergedPrimarySourceIdentifier | None = None, identifier_in_primary_source: str | None = None, stable_target_id: MergedAccessPlatformIdentifier | MergedActivityIdentifier | MergedBibliographicResourceIdentifier | MergedConsentIdentifier | MergedContactPointIdentifier | MergedDistributionIdentifier | MergedOrganizationalUnitIdentifier | MergedOrganizationIdentifier | MergedPersonIdentifier | MergedPrimarySourceIdentifier | MergedResourceIdentifier | MergedVariableGroupIdentifier | MergedVariableIdentifier | None = None) → list[Identity]

Find Identity instances in the in-memory database.

Parameters:

  • had_primary_source – Stable target ID of primary source

  • identifier_in_primary_source – Identifier in the primary source

  • stable_target_id – Stable target ID of the entity

Returns:

List of Identity instances

mex.common.identity.models module

class mex.common.identity.models.Identity(*, identifier: Identifier, hadPrimarySource: MergedPrimarySourceIdentifier, identifierInPrimarySource: str, stableTargetId: Identifier)

Bases: BaseModel

Model for identifier lookup.

hadPrimarySource: Annotated[MergedPrimarySourceIdentifier, FieldInfo(annotation=NoneType, required=True, frozen=True)]
identifier: Annotated[Identifier, FieldInfo(annotation=NoneType, required=True, frozen=True)]
identifierInPrimarySource: Annotated[str, FieldInfo(annotation=NoneType, required=True, frozen=True)]
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]] = {'hadPrimarySource': FieldInfo(annotation=MergedPrimarySourceIdentifier, required=True, frozen=True), 'identifier': FieldInfo(annotation=Identifier, required=True, frozen=True), 'identifierInPrimarySource': FieldInfo(annotation=str, required=True, frozen=True), 'stableTargetId': FieldInfo(annotation=Identifier, required=True, frozen=True)}

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

This replaces Model.__fields__ from Pydantic V1.

stableTargetId: Annotated[Identifier, FieldInfo(annotation=NoneType, required=True, frozen=True)]

mex.common.identity.registry module

mex.common.identity.registry.get_provider() → BaseProvider

Get an instance of the identity provider as configured by identity_provider.

Raises:

RuntimeError – When the configured provider is not registered

Returns:

An instance of a subclass of BaseProvider

mex.common.identity.registry.register_provider(key: Hashable, provider_cls: type[BaseProvider]) → None

Register an implementation of an identity provider to a settings key.

Parameters:

  • key – Possible value of Settings.identity_provider, this will be of type mex.common.identity.types.IdentityProvider on the BaseSettings but maybe overwritten in other packages that have their own settings

  • provider_cls – Implementation of an identity provider

Raises:

RuntimeError – When the key is already registered

Module contents

class mex.common.identity.BaseProvider

Bases: BaseConnector

Base class to define the interface of identity providers.

abstract assign(had_primary_source: MergedPrimarySourceIdentifier, identifier_in_primary_source: str) → Identity

Find an Identity in a database or assign a new one.

abstract fetch(*, had_primary_source: MergedPrimarySourceIdentifier | None = None, identifier_in_primary_source: str | None = None, stable_target_id: MergedAccessPlatformIdentifier | MergedActivityIdentifier | MergedBibliographicResourceIdentifier | MergedConsentIdentifier | MergedContactPointIdentifier | MergedDistributionIdentifier | MergedOrganizationalUnitIdentifier | MergedOrganizationIdentifier | MergedPersonIdentifier | MergedPrimarySourceIdentifier | MergedResourceIdentifier | MergedVariableGroupIdentifier | MergedVariableIdentifier | None = None) → list[Identity]

Find Identity instances matching the given filters.

class mex.common.identity.Identity(*, identifier: Identifier, hadPrimarySource: MergedPrimarySourceIdentifier, identifierInPrimarySource: str, stableTargetId: Identifier)

Bases: BaseModel

Model for identifier lookup.

hadPrimarySource: Annotated[MergedPrimarySourceIdentifier, FieldInfo(annotation=NoneType, required=True, frozen=True)]
identifier: Annotated[Identifier, FieldInfo(annotation=NoneType, required=True, frozen=True)]
identifierInPrimarySource: Annotated[str, FieldInfo(annotation=NoneType, required=True, frozen=True)]
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]] = {'hadPrimarySource': FieldInfo(annotation=MergedPrimarySourceIdentifier, required=True, frozen=True), 'identifier': FieldInfo(annotation=Identifier, required=True, frozen=True), 'identifierInPrimarySource': FieldInfo(annotation=str, required=True, frozen=True), 'stableTargetId': FieldInfo(annotation=Identifier, required=True, frozen=True)}

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

This replaces Model.__fields__ from Pydantic V1.

stableTargetId: Annotated[Identifier, FieldInfo(annotation=NoneType, required=True, frozen=True)]
mex.common.identity.get_provider() → BaseProvider

Get an instance of the identity provider as configured by identity_provider.

Raises:

RuntimeError – When the configured provider is not registered

Returns:

An instance of a subclass of BaseProvider

mex.common.identity.register_provider(key: Hashable, provider_cls: type[BaseProvider]) → None

Register an implementation of an identity provider to a settings key.

Parameters:

  • key – Possible value of Settings.identity_provider, this will be of type mex.common.identity.types.IdentityProvider on the BaseSettings but maybe overwritten in other packages that have their own settings

  • provider_cls – Implementation of an identity provider

Raises:

RuntimeError – When the key is already registered