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 startup 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[Path | str] | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_nested_delimiter: str | None = None, _secrets_dir: str | Path | None = None, *, pdb: bool = False, MEX_SINK: list[Sink] = [Sink.NDJSON], 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 = 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_LDAP_SEARCH_BASE: str = 'DC=rki,DC=local', MEX_WIKI_API_URL: Url = Url('https://wikidata/'), MEX_WEB_USER_AGENT: str = 'rki/mex', MEX_ORCID_API_URL: Url = Url('https://orcid/'), 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
graph_db: str
graph_password: SecretStr
graph_url: str
graph_user: SecretStr
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='pdb', alias_priority=2, validation_alias='MEX_DEBUG', description='Jump into post-mortem debugging after any uncaught exception.'), '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=IdentityProvider, required=False, default=<IdentityProvider.MEMORY: 'memory'>, alias_priority=2, validation_alias='MEX_IDENTITY_PROVIDER', description='Provider to assign identifiers to new model instances.'), 'ldap_search_base': FieldInfo(annotation=str, required=False, default='DC=rki,DC=local', alias_priority=2, validation_alias='MEX_LDAP_SEARCH_BASE', description='Search base for the ldap connector.'), '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='User agent is sent in request headers to external services.'), 'orcid_api_url': FieldInfo(annotation=Url, required=False, default=Url('https://orcid/'), alias_priority=2, validation_alias='MEX_ORCID_API_URL', description='URL of orcid api.'), '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.NDJSON: 'ndjson'>], alias_priority=2, validation_alias='MEX_SINK', description='Where to send data that is extracted or ingested. Defaults to writing ndjson files, but can be configured to push to the backend or the graph.'), '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 the Wikidata API used to resolve an ID to an organization.'), '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.

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)

Bases: Enum

Enum of access level.

READ = 'read'
WRITE = 'write'
class mex.backend.types.DynamicStrEnum(name: str, bases: tuple[type], dct: _EnumDict)

Bases: EnumType

Metaclass to dynamically populate an enumeration from a list of strings.

class mex.backend.types.ExtractedType(value)

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)

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'

Module contents