mex.backend package

Subpackages

Submodules

mex.backend.exceptions module

exception mex.backend.exceptions.BackendError

Bases: MExError

Base backend error that offer details on underlying pydantic errors.

errors() → list[ErrorDetails]

Details about underlying pydantic errors.

class mex.backend.exceptions.DebuggingInfo(*, errors: list[dict[str, Any]], scope: DebuggingScope)

Bases: BaseModel

Debugging information for error responses.

errors: list[dict[str, Any]]
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'errors': FieldInfo(annotation=list[dict[str, Any]], required=True), 'scope': FieldInfo(annotation=DebuggingScope, required=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.

scope: DebuggingScope
class mex.backend.exceptions.DebuggingScope(*, http_version: str, method: str, path: str, path_params: dict[str, Any], query_string: str, scheme: str)

Bases: BaseModel

Scope for debugging info of error responses.

http_version: str
method: str
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'http_version': FieldInfo(annotation=str, required=True), 'method': FieldInfo(annotation=str, required=True), 'path': FieldInfo(annotation=str, required=True), 'path_params': FieldInfo(annotation=dict[str, Any], required=True), 'query_string': FieldInfo(annotation=str, required=True), 'scheme': FieldInfo(annotation=str, required=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.

path: str
path_params: dict[str, Any]
query_string: str
scheme: str
class mex.backend.exceptions.DetailedError(*args, **kwargs)

Bases: Protocol

Protocol for errors that offer details.

errors() → list[ErrorDetails]

Details about each underlying error.

class mex.backend.exceptions.ErrorResponse(*, message: str, debug: DebuggingInfo)

Bases: BaseModel

Response model for user and system errors.

debug: DebuggingInfo
message: str
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'debug': FieldInfo(annotation=DebuggingInfo, required=True), 'message': FieldInfo(annotation=str, required=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.

mex.backend.exceptions.handle_detailed_error(request: Request, exc: Exception) → Response

Handle detailed errors and provide debugging info.

mex.backend.exceptions.handle_uncaught_exception(request: Request, exc: Exception) → Response

Handle uncaught errors and provide debugging info.

mex.backend.fields module

mex.backend.logging module

mex.backend.main module

mex.backend.main.lifespan(_: FastAPI) → AsyncIterator[None]

Async context manager to execute setup and teardown of the FastAPI app.

mex.backend.security module

mex.backend.security._check_header_for_authorization_method(api_key: Annotated[str | None, Depends(APIKeyHeader)] = None, credentials: Annotated[HTTPBasicCredentials | None, Depends(HTTPBasic)] = None, user_agent: Annotated[str, Header(PydanticUndefined)] = 'n/a') → None

Check authorization header for API key or credentials.

Raises:

HTTPException if both API key and credentials or none of them are in header. –

Parameters:

  • api_key – the API key

  • credentials – username and password

  • user_agent – user-agent (in case of a web browser starts with “Mozilla/”)

mex.backend.security.has_read_access(api_key: Annotated[str | None, Depends(APIKeyHeader)] = None, credentials: Annotated[HTTPBasicCredentials | None, Depends(HTTPBasic)] = None, user_agent: Annotated[str, Header(PydanticUndefined)] = 'n/a') → None

Verify if api key or credentials have read access or write access.

Raises:

HTTPException if no header or provided APIKey/credentials have no read access. –

Parameters:

  • api_key – the API key

  • credentials – username and password

  • user_agent – user-agent (in case of a web browser starts with “Mozilla/”)

Settings:

check credentials in backend_user_database or backend_api_key_database

mex.backend.security.has_write_access(api_key: Annotated[str | None, Depends(APIKeyHeader)] = None, credentials: Annotated[HTTPBasicCredentials | None, Depends(HTTPBasic)] = None, user_agent: Annotated[str, Header(PydanticUndefined)] = 'n/a') → None

Verify if provided api key or credentials have write access.

Raises:

HTTPException if no header or provided APIKey/credentials have no write access. –

Parameters:

  • api_key – the API key

  • credentials – username and password

  • user_agent – user-agent (in case of a web browser starts with “Mozilla/”)

Settings:

check credentials in backend_user_database or backend_api_key_database

mex.backend.settings module

class mex.backend.settings.BackendSettings(_env_file: Path | str | Sequence[str | Path] | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_nested_delimiter: str | None = None, _secrets_dir: str | Path | None = None, *, reload: bool = False, MEX_SINK: list[Sink] = [Sink.GRAPH], MEX_ASSETS_DIR: Path = PosixPath('/home/runner/work/mex-backend/mex-backend/assets'), MEX_WORK_DIR: Path = PosixPath('/home/runner/work/mex-backend/mex-backend'), MEX_IDENTITY_PROVIDER: IdentityProvider | BackendIdentityProvider = IdentityProvider.MEMORY, MEX_BACKEND_API_URL: Url = Url('http://localhost:8080/'), MEX_BACKEND_API_KEY: SecretStr = SecretStr('**********'), MEX_VERIFY_SESSION: bool | AssetsPath = True, MEX_ORGANIGRAM_PATH: AssetsPath = AssetsPath('raw-data/organigram/organizational_units.json'), MEX_PRIMARY_SOURCES_PATH: AssetsPath = AssetsPath('raw-data/primary-sources/primary-sources.json'), MEX_LDAP_URL: SecretStr = SecretStr('**********'), MEX_WIKI_API_URL: Url = Url('https://wikidata/'), MEX_WIKI_QUERY_SERVICE_URL: Url = Url('https://wikidata/'), MEX_WEB_USER_AGENT: str = 'rki/mex', MEX_BACKEND_HOST: Annotated[str, MinLen(min_length=1), MaxLen(max_length=250)] = 'localhost', MEX_BACKEND_PORT: Annotated[int, Gt(gt=0), Lt(lt=65536)] = 8080, MEX_BACKEND_ROOT_PATH: str = '', MEX_GRAPH_URL: str = 'neo4j://localhost:7687', MEX_GRAPH_NAME: str = 'neo4j', MEX_GRAPH_USER: SecretStr = SecretStr('**********'), MEX_GRAPH_PASSWORD: SecretStr = SecretStr('**********'), MEX_BACKEND_API_KEY_DATABASE: APIKeyDatabase = APIKeyDatabase(read=[], write=[]), MEX_BACKEND_API_USER_DATABASE: APIUserDatabase = APIUserDatabase(read={}, write={}))

Bases: BaseSettings

Settings definition for the backend server.

backend_api_key_database: APIKeyDatabase
backend_host: str
backend_port: int
backend_root_path: str
backend_user_database: APIUserDatabase
debug: bool
graph_db: str
graph_password: SecretStr
graph_url: str
graph_user: SecretStr
identity_provider: IdentityProvider | BackendIdentityProvider
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'mex_', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'populate_by_name': True, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_assignment': True, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'assets_dir': FieldInfo(annotation=Path, required=False, default=PosixPath('/home/runner/work/mex-backend/mex-backend/assets'), alias_priority=2, validation_alias='MEX_ASSETS_DIR', description='Path to directory that contains input files treated as read-only, looks for a folder named `assets` in the current directory by default.'), 'backend_api_key': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_BACKEND_API_KEY', description='Backend API key with write access to call POST/PUT endpoints'), 'backend_api_key_database': FieldInfo(annotation=APIKeyDatabase, required=False, default=APIKeyDatabase(read=[], write=[]), alias_priority=2, validation_alias='MEX_BACKEND_API_KEY_DATABASE', description='Database of API keys.'), 'backend_api_url': FieldInfo(annotation=Url, required=False, default=Url('http://localhost:8080/'), alias_priority=2, validation_alias='MEX_BACKEND_API_URL', description='MEx backend API url.'), 'backend_host': FieldInfo(annotation=str, required=False, default='localhost', alias_priority=2, validation_alias='MEX_BACKEND_HOST', description='Host that the backend server will run on.', metadata=[MinLen(min_length=1), MaxLen(max_length=250)]), 'backend_port': FieldInfo(annotation=int, required=False, default=8080, alias_priority=2, validation_alias='MEX_BACKEND_PORT', description='Port that the backend server should listen on.', metadata=[Gt(gt=0), Lt(lt=65536)]), 'backend_root_path': FieldInfo(annotation=str, required=False, default='', alias_priority=2, validation_alias='MEX_BACKEND_ROOT_PATH', description='Root path that the backend server should run under.'), 'backend_user_database': FieldInfo(annotation=APIUserDatabase, required=False, default=APIUserDatabase(read={}, write={}), alias_priority=2, validation_alias='MEX_BACKEND_API_USER_DATABASE', description='Database of users.'), 'debug': FieldInfo(annotation=bool, required=False, default=False, alias='reload', alias_priority=2, validation_alias='MEX_DEBUG', description='Enable debug mode.'), 'graph_db': FieldInfo(annotation=str, required=False, default='neo4j', alias_priority=2, validation_alias='MEX_GRAPH_NAME', description='Name of the default graph database.'), 'graph_password': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_GRAPH_PASSWORD', description='Password for authenticating with the graph database.'), 'graph_url': FieldInfo(annotation=str, required=False, default='neo4j://localhost:7687', alias_priority=2, validation_alias='MEX_GRAPH_URL', description='URL for connecting to the graph database.'), 'graph_user': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_GRAPH_USER', description='Username for authenticating with the graph database.'), 'identity_provider': FieldInfo(annotation=Union[IdentityProvider, BackendIdentityProvider], required=False, default=<IdentityProvider.MEMORY: 'memory'>, alias_priority=2, validation_alias='MEX_IDENTITY_PROVIDER', description='Provider to assign stableTargetIds to new model instances.'), 'ldap_url': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_LDAP_URL', description='LDAP server for person queries with authentication credentials. Must follow format `ldap://user:pw@host:port`, where `user` is the username, and `pw` is the password for authenticating against ldap, `host` is the url of the ldap server, and `port` is the port of the ldap server.'), 'mex_web_user_agent': FieldInfo(annotation=str, required=False, default='rki/mex', alias_priority=2, validation_alias='MEX_WEB_USER_AGENT', description='a user agent is sent in the header of some requests to external services '), 'organigram_path': FieldInfo(annotation=AssetsPath, required=False, default=AssetsPath("raw-data/organigram/organizational_units.json"), alias_priority=2, validation_alias='MEX_ORGANIGRAM_PATH', description='Path to the JSON file describing the organizational units, absolute path or relative to `assets_dir`.'), 'primary_sources_path': FieldInfo(annotation=AssetsPath, required=False, default=AssetsPath("raw-data/primary-sources/primary-sources.json"), alias_priority=2, validation_alias='MEX_PRIMARY_SOURCES_PATH', description='Path to the JSON file describing the primary sources, absolute path or relative to `assets_dir`.'), 'sink': FieldInfo(annotation=list[Sink], required=False, default=[<Sink.GRAPH: 'graph'>], alias_priority=2, validation_alias='MEX_SINK', description='Where to send ingested data. Defaults to writing to the graph db.'), 'verify_session': FieldInfo(annotation=Union[bool, AssetsPath], required=False, default=True, alias_priority=2, validation_alias='MEX_VERIFY_SESSION', description="Either a boolean that controls whether we verify the server's TLS certificate, or a path to a CA bundle to use. If a path is given, it can be either absolute or relative to the `assets_dir`. Defaults to True."), 'wiki_api_url': FieldInfo(annotation=Url, required=False, default=Url('https://wikidata/'), alias_priority=2, validation_alias='MEX_WIKI_API_URL', description='URL of Wikidata API, this URL is used to send wikidata organization ID to get all the info about the organization, which includes basic info, aliases, labels, descriptions, claims, and sitelinks'), 'wiki_query_service_url': FieldInfo(annotation=Url, required=False, default=Url('https://wikidata/'), alias_priority=2, validation_alias='MEX_WIKI_QUERY_SERVICE_URL', description='URL of Wikidata query service, this URL is to send organization name in plain text to wikidata and receive search results with wikidata organization ID'), 'work_dir': FieldInfo(annotation=Path, required=False, default=PosixPath('/home/runner/work/mex-backend/mex-backend'), alias_priority=2, validation_alias='MEX_WORK_DIR', description='Path to directory that stores generated and temporary files. Defaults to the current working directory.')}

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.

sink: list[Sink]

mex.backend.types module

class mex.backend.types.APIKey(secret_value: SecretType)

Bases: SecretStr

An API Key used for authenticating and authorizing a client.

class mex.backend.types.APIKeyDatabase(*, read: list[APIKey] = [], write: list[APIKey] = [])

Bases: BaseModel

A lookup from access level to list of allowed APIKeys.

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]] = {'read': FieldInfo(annotation=list[APIKey], required=False, default=[]), 'write': FieldInfo(annotation=list[APIKey], required=False, default=[])}

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.

read: list[APIKey]
write: list[APIKey]
class mex.backend.types.APIUserDatabase(*, read: dict[str, APIUserPassword] = {}, write: dict[str, APIUserPassword] = {})

Bases: BaseModel

Database containing usernames and passwords for backend API.

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]] = {'read': FieldInfo(annotation=dict[str, APIUserPassword], required=False, default={}), 'write': FieldInfo(annotation=dict[str, APIUserPassword], required=False, default={})}

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.

read: dict[str, APIUserPassword]
write: dict[str, APIUserPassword]
class mex.backend.types.APIUserPassword(secret_value: SecretType)

Bases: SecretStr

An API password used for basic authentication along with a username.

class mex.backend.types.AccessLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enum of access level.

READ = 'read'
WRITE = 'write'
class mex.backend.types.BackendIdentityProvider(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Identity providers implemented by mex-backend.

GRAPH = 'graph'
class mex.backend.types.DynamicStrEnum(name: str, bases: tuple[type], dct: _EnumDict)

Bases: EnumType

Meta class to dynamically populate the an enumeration from a list of strings.

class mex.backend.types.ExtractedType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of possible types for extracted items.

EXTRACTED_ACCESS_PLATFORM = 'ExtractedAccessPlatform'
EXTRACTED_ACTIVITY = 'ExtractedActivity'
EXTRACTED_BIBLIOGRAPHIC_RESOURCE = 'ExtractedBibliographicResource'
EXTRACTED_CONTACT_POINT = 'ExtractedContactPoint'
EXTRACTED_DISTRIBUTION = 'ExtractedDistribution'
EXTRACTED_ORGANIZATION = 'ExtractedOrganization'
EXTRACTED_ORGANIZATIONAL_UNIT = 'ExtractedOrganizationalUnit'
EXTRACTED_PERSON = 'ExtractedPerson'
EXTRACTED_PRIMARY_SOURCE = 'ExtractedPrimarySource'
EXTRACTED_RESOURCE = 'ExtractedResource'
EXTRACTED_VARIABLE = 'ExtractedVariable'
EXTRACTED_VARIABLE_GROUP = 'ExtractedVariableGroup'
class mex.backend.types.MergedType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of possible types for merged items.

MERGED_ACCESS_PLATFORM = 'MergedAccessPlatform'
MERGED_ACTIVITY = 'MergedActivity'
MERGED_BIBLIOGRAPHIC_RESOURCE = 'MergedBibliographicResource'
MERGED_CONTACT_POINT = 'MergedContactPoint'
MERGED_DISTRIBUTION = 'MergedDistribution'
MERGED_ORGANIZATION = 'MergedOrganization'
MERGED_ORGANIZATIONAL_UNIT = 'MergedOrganizationalUnit'
MERGED_PERSON = 'MergedPerson'
MERGED_PRIMARY_SOURCE = 'MergedPrimarySource'
MERGED_RESOURCE = 'MergedResource'
MERGED_VARIABLE = 'MergedVariable'
MERGED_VARIABLE_GROUP = 'MergedVariableGroup'

mex.backend.utils module

mex.backend.utils.extend_list_in_dict(dict_: dict[str, list[T]], key: str, item: list[T] | T) → None

Extend a list in a dict for a given key with the given unique item(s).

mex.backend.utils.prune_list_in_dict(dict_: dict[str, list[T]], key: str, item: list[T] | T) → None

Safely remove item(s) from a list in a dict for the given key.

Module contents