mex.drop.api package

Submodules

mex.drop.api.main module

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

Bases: BaseModel

Response model for system status check.

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]] = {'status': FieldInfo(annotation=str, required=True), 'version': 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.

status: str
version: str
mex.drop.api.main.check_system_status() → SystemStatus

Check that the drop server is healthy and responsive.

async mex.drop.api.main.download_data(x_system: Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')], Path(PydanticUndefined)], entity_type: Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')], Path(PydanticUndefined)], authorized_x_systems: Annotated[list[Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')]]], Depends(get_current_authorized_x_systems)]) → Response

Download data from MEx.

Parameters:

  • x_system – name of the x-system that is the original data source

  • entity_type – name of the data file to download

  • authorized_x_systems – list of authorized x-systems

Settings:

drop_directory: where data is stored

Returns:

A response

async mex.drop.api.main.drop_data(x_system: Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')], Path(PydanticUndefined)], entity_type: Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')], Path(PydanticUndefined)], data: Annotated[dict[str, Any] | list[Any] | bytes, Body(PydanticUndefined)], content_type: Annotated[str, Header(PydanticUndefined)], authorized_x_systems: Annotated[list[Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')]]], Depends(get_current_authorized_x_systems)]) → Response

Upload arbitrary 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 – data content of request body

  • authorized_x_systems – list of authorized x-systems

  • content_type – Content-Type of the uploaded data

Settings:

drop_directory: where accepted data is stored

Returns:

A JSON response

async mex.drop.api.main.drop_data_multipart(x_system: Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')], Path(PydanticUndefined)], files: Annotated[list[UploadFile], File(PydanticUndefined)], authorized_x_systems: Annotated[list[Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')]]], Depends(get_current_authorized_x_systems)], background_tasks: BackgroundTasks) → Response

Upload multipart data to MEx.

Parameters:

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

  • files – list of files to be uploaded to MEx

  • authorized_x_systems – list of authorized x-systems

  • background_tasks – collection of background tasks

Settings:

drop_directory: where accepted data is stored

Returns:

A JSON response

mex.drop.api.main.list_entity_types(x_system: Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')], Path(PydanticUndefined)], authorized_x_systems: Annotated[list[Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')]]], Depends(get_current_authorized_x_systems)]) → dict[str, list[str]]

List available files for an x-system.

Parameters:

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

  • authorized_x_systems – list of authorized x-systems

Settings:

drop_directory: where data is stored

Returns:

A response

mex.drop.api.main.list_x_systems(authorized_x_systems: Annotated[list[Annotated[str, Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\._-]{1,128}$')]]], Depends(get_current_authorized_x_systems)]) → dict[str, list[str]]

List x-systems with available data.

Parameters:

authorized_x_systems – list of authorized x-systems

Settings:

drop_directory: where data is stored

Returns:

A response

Module contents