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