mex.drop package

Subpackages

• mex.drop.sinks package
  • Submodules
  • mex.drop.sinks.json module
    • json_sink()
  • Module contents

Submodules

mex.drop.logging module

mex.drop.main module

class mex.drop.main.SystemStatus(*, status: str)

Bases: BaseModel

Response model for system status check.

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[dict[str, FieldInfo]] = {'status': FieldInfo(annotation=str, required=True)}

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

This replaces Model.__fields__ from Pydantic V1.

status: str

mex.drop.main.check_system_status() → SystemStatus

Check that the drop server is healthy and responsive.

async mex.drop.main.drop_data(x_system: str, entity_type: str, data: dict[str, Any] | list[Any], x_systems: list[str]) → Response

Upload arbitrary JSON data to MEx.

Parameters:

• x_system – name of the x-system that the data comes from

• entity_type – name of the data file that is uploaded, if unsure use ‘default’

• data – dictionary with string keys or list with and arbitrary values

• x_systems – list of authorized x-systems

Settings:

drop_directory: where accepted data is stored

Returns:

A JSON response

mex.drop.main.show_form(request: Request, x_system: str, entity_type: str) → Response

Render an HTML upload form for easier data dropping.

Parameters:

• request – the incoming request

• x_system – name of the x-system that the data comes from

• entity_type – name of the data file that is uploaded, if unsure use ‘default’

Returns:

An HTML response

mex.drop.security module

mex.drop.security.get_authorized_x_systems(user_database: UserDatabase, api_key: APIKey) → list[str] | None

Get authorized x-systems from database.

Parameters:

• user_database – database dictionary

• api_key – API key to check

Returns:

optional list of authorized x-systems

mex.drop.security.get_current_authorized_x_systems(api_key: str | None) → list[str]

Get the current authorized x-systems.

Raises:

HTTPException if no header is provided or no x-system is authorized –

Parameters:

api_key – the API key for x-system lookup

Settings:

drop_user_database: checked for presence of api_key

Returns:

list of authorized x-systems

mex.drop.settings module

class mex.drop.settings.DropSettings(_env_file: Path | str | List[Path | str] | Tuple[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, sink: list[Sink] = [Sink.NDJSON], assets_dir: Path = PosixPath('/home/runner/work/mex-drop/mex-drop/assets'), work_dir: Path = PosixPath('/home/runner/work/mex-drop/mex-drop'), identity_provider: IdentityProvider = IdentityProvider.MEMORY, backend_api_url: Url = Url('http://localhost:8080/'), backend_api_key: SecretStr = SecretStr('**********'), verify_session: bool | AssetsPath = True, public_api_url: Url = Url('http://localhost:53000/'), public_api_token_provider: Url = Url('http://localhost:53000/api/v0/oauth/token'), public_api_token_payload: SecretStr = SecretStr('**********'), public_api_verify_session: bool | AssetsPath = True, organigram_path: AssetsPath = AssetsPath('raw-data/organigram/organizational_units.json'), primary_sources_path: AssetsPath = AssetsPath('raw-data/primary-sources/primary-sources.json'), ldap_url: SecretStr = SecretStr('**********'), wiki_api_url: Url = Url('https://wikidata/'), wiki_query_service_url: Url = Url('https://wikidata/'), drop_host: str = 'localhost', drop_port: int = 8081, drop_root_path: str = '', drop_directory: WorkPath = WorkPath('data'), drop_user_database: UserDatabase = {})

Bases: BaseSettings

Settings definition for the drop server.

drop_directory: WorkPath

drop_host: str

drop_port: int

drop_root_path: str

drop_user_database: UserDatabase

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_nested_delimiter': None, 'env_prefix': 'mex_', 'extra': 'ignore', 'populate_by_name': True, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, '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]] = {'assets_dir': FieldInfo(annotation=Path, required=False, default=PosixPath('/home/runner/work/mex-drop/mex-drop/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_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.'), '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.'), 'drop_directory': FieldInfo(annotation=WorkPath, required=False, default=WorkPath("data"), alias_priority=2, validation_alias='MEX_DROP_DIRECTORY', description='Root directory that the drop server should save files in, absolute or relative to `work_dir`.'), 'drop_host': FieldInfo(annotation=str, required=False, default='localhost', alias_priority=2, validation_alias='MEX_DROP_HOST', description='Host that the drop server will run on.', metadata=[MinLen(min_length=1), MaxLen(max_length=250)]), 'drop_port': FieldInfo(annotation=int, required=False, default=8081, alias_priority=2, validation_alias='MEX_DROP_PORT', description='Port that the drop server should listen on.', metadata=[Gt(gt=0), Lt(lt=65536)]), 'drop_root_path': FieldInfo(annotation=str, required=False, default='', alias_priority=2, validation_alias='MEX_DROP_ROOT_PATH', description='Root path that the drop server should run under.'), 'drop_user_database': FieldInfo(annotation=UserDatabase, required=False, default={}, alias_priority=2, validation_alias='MEX_DROP_USER_DATABASE', description='Database of users.'), 'identity_provider': FieldInfo(annotation=IdentityProvider, 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.'), '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`.'), 'public_api_token_payload': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_PUBLIC_API_TOKEN_PAYLOAD', description='Base64-encoded payload to send when requesting a JWT for the public API.'), 'public_api_token_provider': FieldInfo(annotation=Url, required=False, default=Url('http://localhost:53000/api/v0/oauth/token'), alias_priority=2, validation_alias='MEX_PUBLIC_API_TOKEN_PROVIDER', description='URL of the JSON Web Token provider for the public API.'), 'public_api_url': FieldInfo(annotation=Url, required=False, default=Url('http://localhost:53000/'), alias_priority=2, validation_alias='MEX_PUBLIC_API_URL', description='MEx public API url.'), 'public_api_verify_session': FieldInfo(annotation=Union[bool, AssetsPath], required=False, default=True, alias_priority=2, validation_alias='MEX_PUBLIC_API_VERIFY_SESSION', description='Public API-specific session verification setting, see `verify_session` for possible values.'), '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 set to backend or public APIs or to 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 organizatizion 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-drop/mex-drop'), 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].

This replaces Model.__fields__ from Pydantic V1.

classmethod validate_user_database(value: dict[str, list[str]]) → UserDatabase

Ensure keys are APIKeys and values are XSystems.

mex.drop.types module

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

Bases: SecretStr

An API Key used for authenticating and authorizing a client.

class mex.drop.types.UserDatabase

Bases: dict[APIKey, list[str]]

A lookup of which x-systems each APIKey is allowed to access.

Module contents