py-tes 1.1.0rc3

Library for communicating with the GA4GH Task Execution API

py-tes 🐍

py-tes is a library for interacting with servers implementing the GA4GH Task Execution Schema.

Quick Start ⚡

TES version	py-tes version	Example Notebook
1.1.0 (latest)	1.1.0
1.0.0	1.0.0

Installation 🌀

Install py-tes from PyPI and run it in your script:

➜ pip install py-tes

➜ python example.py

example.py 🐍

import tes
import json

# Define task
task = tes.Task(
    executors=[
        tes.Executor(
            image="alpine",
            command=["echo", "hello"]
        )
    ]
)

# Create client
cli = tes.HTTPClient("http://localhost:8000", timeout=5)

# Create and run task
task_id = cli.create_task(task)
cli.wait(task_id, timeout=5)

# Fetch task info
task_info = cli.get_task(task_id, view="BASIC")
j = json.loads(task_info.as_json())

# Pretty print task info
print(json.dumps(j, indent=2))

How to...

Makes use of the objects above...

...export a model to a dictionary

task_dict = task.as_dict(drop_empty=False)

task_dict contents:

{'id': None, 'state': None, 'name': None, 'description': None, 'inputs': None, 'outputs': None, 'resources': None, 'executors': [{'image': 'alpine', 'command': ['echo', 'hello'], 'workdir': None, 'stdin': None, 'stdout': None, 'stderr': None, 'env': None}], 'volumes': None, 'tags': None, 'logs': None, 'creation_time': None}

...export a model to JSON

task_json = task.as_json()  # also accepts `drop_empty` arg

task_json contents:

{"executors": [{"image": "alpine", "command": ["echo", "hello"]}]}

...pretty print a model

print(task.as_json(indent=3))  # keyword args are passed to `json.dumps()`

Output:

{
  "executors": [
    {
      "image": "alpine",
      "command": ["echo", "hello"]
    }
  ]
}

...access a specific task from the task list

specific_task = tasks_list.tasks[5]

specific_task contents:

Task(id='393K43', state='COMPLETE', name=None, description=None, inputs=None, outputs=None, resources=None, executors=None, volumes=None, tags=None, logs=None, creation_time=None)

...iterate over task list items

for t in tasks_list[:3]:
    print(t.as_json(indent=3))

Output:

{
   "id": "task_A2GFS4",
   "state": "RUNNING"
}
{
   "id": "task_O8G1PZ",
   "state": "CANCELED"
}
{
   "id": "task_W246I6",
   "state": "COMPLETE"
}

...instantiate a model from a JSON representation

task_from_json = tes.client.unmarshal(task_json, tes.Task)

task_from_json contents:

Task(id=None, state=None, name=None, description=None, inputs=None, outputs=None, resources=None, executors=[Executor(image='alpine', command=['echo', 'hello'], workdir=None, stdin=None, stdout=None, stderr=None, env=None)], volumes=None, tags=None, logs=None, creation_time=None)

Which is equivalent to task:

print(task_from_json == task)

Output:

True

Additional Resources 📚

• ga4gh-tes : C# implementation of the GA4GH TES API; provides distributed batch task execution on Microsoft Azure

• cwl-tes : cwl-tes submits your tasks to a TES server. Task submission is parallelized when possible.

• Funnel: Funnel is a toolkit for distributed task execution with a simple API.

• Snakemake : The Snakemape workflow management system is a tool to create reproducible and scalable data analyses

• Nextflow: Nextflow enables scalable and reproducible scientific workflows using software containers. It allows the adaptation of pipelines written in the most common scripting languages.

• GA4GH TES: Main page for the Task Execution Schema — a standardized schema and API for describing batch execution tasks.

• TES GitHub: Source repo for the Task Execution Schema

• Awesome TES: A curated list of awesome GA4GH TES projects and programs