py-tes

tes.client module

TES access methods and helper functions.

class tes.client.HTTPClient(url, timeout: int = 10, user: Any = None, password: Any = None, token: Any = None)[source]

Bases: object

HTTP client class for interacting with the TES API.

cancel_task(task_id: str) → None[source]

Access method for POST /tasks/{id}:cancel.

Parameters:

task_id – TES Task ID.

create_task(task: Task) → CreateTaskResponse[source]

Access method for POST /tasks.

Parameters:

task – tes.models.Task instance.

Returns:

tes.models.CreateTaskResponse instance.

Raises:

TypeError – If task is not a tes.models.Task instance.

get_service_info() → ServiceInfo[source]

Access method for GET /service-info.

Returns:

tes.models.ServiceInfo instance.

get_task(task_id: str, view: str = 'BASIC') → Task[source]

Access method for GET /tasks/{id}.

Parameters:

  • task_id – TES Task ID.

  • view – Task info verbosity. One of MINIMAL, BASIC and FULL.

Returns:

tes.models.Task instance.

list_tasks(view: str = 'MINIMAL', page_size: int | None = None, page_token: str | None = None) → ListTasksResponse[source]

Access method for GET /tasks.

Parameters:

  • view – Task info verbosity. One of MINIMAL, BASIC and FULL.

  • page_size – Number of tasks to return.

  • page_token – Token to retrieve the next page of tasks.

Returns:

tes.models.ListTasksResponse instance.

password: str | None
timeout: int
token: str | None
url: str
user: str | None
wait(task_id: str, timeout=None) → Task[source]
tes.client.append_suffixes_to_url(urls: List[str], suffixes: List[str]) → List[str][source]

Compile all combinations of full paths from paths and suffixes.

Parameters:

  • urls – List of URL paths.

  • prefixes – List of suffixes to be appended to urls.

Returns:

List of full path combinations, in the provided order of paths and suffixes, starting with all suffix combinations for the first path, then those for the second path, and so on. Paths are stripped of any trailing slashes.

Examples

>>> client = tes.HTTPClient.append_suffixes_to_url(['https://funnel.exa
mple.com'], ['ga4gh/tes/v1', 'v1', ''])
['https://funnel.example.com/ga4gh/tes/v1', 'https://funnel.example.com
/v1', 'https://funnel.example.com']
tes.client.process_url(value)[source]
tes.client.send_request(paths: List[str], method: str = 'get', kwargs_requests: Dict[str, Any] | None = None, **kwargs: Any) → Response[source]

Send request to a list of URLs, returning the first valid response.

Parameters:

  • paths – List of fully qualified URLs.

  • method – HTTP method to use for the request; one of ‘get’ (default), ‘post’, ‘put’, and ‘delete’.

  • kwargs_requests – Keyword arguments to pass to the requests call.

  • **kwargs – Keyword arguments for path parameter substition.

Returns:

The first successful response from the list of endpoints.

Raises:

  • requests.exceptions.HTTPError – If no response is received from any path.

  • requests.exceptions.HTTPError – As soon as the first 4xx or 5xx status code is received.

  • requests.exceptions.HTTPError – If, after trying all paths, at least one 404 status code and no other 4xx or 5xx status codes are received.

  • ValueError – If an unsupported HTTP method is provided.

tes.models module

TES models, converters, validators and helpers.

class tes.models.Base[source]

Bases: object

attrs base class for all TES and helper models.

as_dict(drop_empty: bool = True) → Dict[str, Any][source]
as_json(drop_empty: bool = True, **kwargs) → str[source]
class tes.models.CancelTaskRequest(id: Any)[source]

Bases: Base

attrs model class for POST /tasks/{id}:cancel request parameters.

id: str
class tes.models.CancelTaskResponse[source]

Bases: Base

TES tesCancelTaskResponse attrs model class.

class tes.models.CreateTaskResponse(id: Any)[source]

Bases: Base

TES tesCreateTaskResponse attrs model class.

id: str
class tes.models.Executor(image: Any, command: Any, ignore_error: Any = None, workdir: Any = None, stdin: Any = None, stdout: Any = None, stderr: Any = None, env: Dict | None = None)[source]

Bases: Base

TES tesExecutor attrs model class.

command: List[str]
env: Dict | None
ignore_error: str
image: str
stderr: str
stdin: str
stdout: str
workdir: str
class tes.models.ExecutorLog(start_time: str | None = None, end_time: str | None = None, stdout: Any = None, stderr: Any = None, exit_code: int | None = None)[source]

Bases: Base

TES tesExecutorLog attrs model class.

end_time: datetime
exit_code: int | None
start_time: datetime
stderr: str | None
stdout: str | None
class tes.models.GetTaskRequest(id: Any, view: str | None = None)[source]

Bases: Base

attrs model class for GET /tasks/{id} request parameters.

id: str
view: str | None
class tes.models.Input(url: Any = None, path: Any = None, type: str = 'FILE', name: Any = None, description: Any = None, content: Any = None, streamable: bool | None = None)[source]

Bases: Base

TES tesInput attrs model class.

content: str | None
description: str | None
name: str | None
path: str | None
streamable: bool | None
type: str
url: str | None
class tes.models.ListTasksRequest(project: Any = None, name_prefix: Any = None, page_size: int | None = None, page_token: Any = None, view: str | None = None)[source]

Bases: Base

attrs model class for GET /tasks request parameters.

name_prefix: str | None
page_size: int | None
page_token: str | None
project: str | None
view: str | None
class tes.models.ListTasksResponse(tasks: List[Task] | None = None, next_page_token: Any = None)[source]

Bases: Base

TES tesListTasksResponse attrs model class.

next_page_token: str | None
tasks: List[Task] | None
class tes.models.Organization(name: Any = None, url: Any = None)[source]

Bases: object

name: str | None
url: str | None
class tes.models.Output(url: Any = None, path: Any = None, path_prefix: Any = None, type: str = 'FILE', name: Any = None, description: Any = None)[source]

Bases: Base

TES tesOutput attrs model class.

description: str | None
name: str | None
path: str | None
path_prefix: str | None
type: str
url: str | None
class tes.models.OutputFileLog(url: Any = None, path: Any = None, size_bytes: str | None = None)[source]

Bases: Base

TES tesOutputFileLog attrs model class.

path: str | None
size_bytes: int | None
url: str | None
class tes.models.Resources(cpu_cores: int | None = None, ram_gb: float | int | None = None, disk_gb: float | int | None = None, preemptible: bool | None = None, zones: Any = None, backend_parameters: Dict[str, str] | None = None, backend_parameters_strict: bool | None = None)[source]

Bases: Base

TES tesResources attrs model class.

backend_parameters: Dict[str, str] | None
backend_parameters_strict: bool | None
cpu_cores: int | None
disk_gb: float | int | None
preemptible: bool | None
ram_gb: float | int | None
zones: List[str] | None
class tes.models.ServiceInfo(contact_url: Any = None, created_at: Any = None, description: Any = None, documentation_url: Any = None, environment: Any = None, id: Any = None, name: Any = None, organization: dict | None = None, storage: Any = None, tes_resources_backend_parameters: Any = None, type: dict | None = None, updated_at: Any = None, version: Any = None)[source]

Bases: Base

TES tesServiceInfo attrs model class.

contact_url: str | None
created_at: str | None
description: str | None
documentation_url: str | None
environment: str | None
id: str | None
name: str | None
organization: dict | None
storage: List[str] | None
tes_resources_backend_parameters: List[str] | None
type: dict | None
updated_at: str | None
version: str | None
class tes.models.ServiceInfoRequest[source]

Bases: Base

attrs model class for GET /service-info request parameters.

class tes.models.Task(id: Any = None, state: str | None = None, name: Any = None, description: Any = None, inputs: List[Input] | None = None, outputs: List[Output] | None = None, resources: Resources | None = None, executors: List[Executor] | None = None, volumes: List[str] | None = None, tags: Dict | None = None, logs: List[TaskLog] | None = None, creation_time: str | None = None)[source]

Bases: Base

TES tesTask attrs model class.

creation_time: datetime
description: str | None
executors: List[Executor] | None
id: str | None
inputs: List[Input] | None
is_valid() → Tuple[bool, None | TypeError][source]

Validate a Task model instance.

Returns:

A tuple containing a boolean indicating whether the model is valid, and a TypeError if the model is invalid, or None if it is.

logs: List[TaskLog] | None
name: str | None
outputs: List[Output] | None
resources: Resources | None
state: str | None
tags: Dict | None
volumes: List[str] | None
class tes.models.TaskLog(start_time: str | None = None, end_time: str | None = None, metadata: Dict | None = None, logs: List[ExecutorLog] | None = None, outputs: List[OutputFileLog] | None = None, system_logs: List[str] | None = None)[source]

Bases: Base

TES tesTaskLog attrs model class.

end_time: datetime
logs: List[ExecutorLog] | None
metadata: Dict | None
outputs: List[OutputFileLog] | None
start_time: datetime
system_logs: List[str] | None
class tes.models.Type(artifact: Any = None, group: Any = None, version: Any = None)[source]

Bases: object

artifact: str | None
group: str | None
version: str | None
tes.models.datetime_json_handler(x: Any) → str[source]

JSON handler for datetime objects.

Parameters:

x – Object to convert.

Returns:

Converted object.

Raises:

TypeError – If x is not a datetime object.

tes.models.int64conv(value: str | None) → int | None[source]

Convert string to int64.

Parameters:

value – String to convert.

Returns:

Converted value.

tes.models.list_of(_type: Any) → _ListOfValidator[source]

attrs validator for lists of a given type.

Parameters:

_type – Type to validate.

Returns:

attrs validator for the given type.

tes.models.strconv(value: Any) → Any[source]

Explicitly cast a string-like value or list thereof to string(s).

Parameters:

value – Value to convert.

Returns:

Converted value. If value is a list, all elements are converted to strings. If value is not string-like, it will be returned as is.

tes.models.timestampconv(value: str | None) → datetime | None[source]

Convert string to datetime.

Parameters:

value – String to convert.

Returns:

Converted value.

tes.utils module

Exceptions and utilities.

exception tes.utils.TimeoutError(*args, **kwargs)[source]

Bases: Exception

exception tes.utils.UnmarshalError(*args, **kwargs)[source]

Bases: Exception

Raised when a JSON string cannot be unmarshalled to a TES model.

tes.utils.camel_to_snake(name: str) → str[source]

Converts camelCase to snake_case.

Parameters:

name – String to convert.

Returns:

Converted string.

tes.utils.unmarshal(j: Any, o: Type, convert_camel_case=True) → Any[source]

Unmarshal a JSON string to a TES model.

Parameters:

  • j – JSON string or dictionary to unmarshal.

  • o – TES model to unmarshal to.

  • convert_camel_case – Convert values in j from camelCase to snake_case.

Returns:

Unmarshalled TES model.

Raises:

UnmarshalError – If j cannot be unmarshalled to o.