PyUNICORE, a Python library for using UNICORE and UFTP

This library covers the UNICORE REST API, making common tasks like file access, job submission and management, workflow submission and management more convenient, and integrating UNICORE features better with typical Python usage. Both blocking and non-blocking (asyncio) styles of communication with the server are supported.

The full, up-to-date documentation of the REST API can be found here

In addition, this library contains code for using UFTP (UNICORE FTP) for filesystem mounts with FUSE, a UFTP driver for PyFilesystem and a UNICORE implementation of a Dask Cluster

PyUNICORE comes with a commandline utility 'unicore', which is modeled after the UNICORE Commandline Client (UCC) and supports many of UCC's features.

This project has received funding from the European Union’s Horizon 2020 Framework Programme for Research and Innovation under the Specific Grant Agreement Nos. 720270, 785907 and 945539 (Human Brain Project SGA 1, 2 and 3)

See LICENSE file for licensing information

Documentation

The complete documentation of PyUNICORE can be viewed here

Installation

Install from PyPI with

pip install -U pyunicore

Additional extra packages may be required for your use case:

• Creating JWT tokens signed with keys requires the "cryptography" package
• Using UFTP with pyfilesystem requires "fs"

You can install (one or more) extras with pip:

pip install -U pyunicore[crypto,fs,fuse]

Basic usage

Creating a client for a UNICORE site

import pyunicore.client as uc_client
import pyunicore.credentials as uc_credentials
import json

base_url = "https://localhost:8080/DEMO-SITE/rest/core"

# authenticate with username/password
credential = uc_credentials.UsernamePassword("demouser", "test123")

client = uc_client.Client(credential, base_url)
print(json.dumps(client.properties, indent = 2))

PyUNICORE supports a variety of authentication options.

Run a job and read result files

my_job = {'Executable': 'date'}

job = client.new_job(job_description=my_job, inputs=[])
print(json.dumps(job.properties, indent = 2))

job.poll() # wait for job to finish

work_dir = job.working_dir
print(json.dumps(work_dir.properties, indent = 2))

stdout = work_dir.stat("/stdout")
print(json.dumps(stdout.properties, indent = 2))
content = stdout.read()
print(content)

Connect to a Registry and list all registered services

registry_url = "https://localhost:8080/REGISTRY/rest/registries/default_registry"

# authenticate with username/password
credential = uc_credentials.UsernamePassword("demouser", "test123")

r = uc_client.Registry(credential, registry_url)
print(r.site_urls)

Further reading

More examples for using PyUNICORE can be found in the "integration-tests" folder in the source code repository.

UFTP examples

Using UFTP for PyFilesystem

You can create a PyFilesystem FS object either directly in code, or implicitely via a URL.

The convenient way is via URL:

from fs import open_fs
fs_url = "uftp://demouser:test123@localhost:9000/rest/auth/TEST:/data"
uftp_fs = open_fs(fs_url)

More...

Tunneling / port forwarding

Opens a local server socket for clients to connect to, where traffic gets forwarded to a service on a HPC cluster login (or compute) node. This feature requires UNICORE 9.1.0 or later on the server side.

You can use this feature in two ways

• in your own applications via the pyunicore.client.Job class.
• you can also open a tunnel from the command line using the 'pyunicore.forwarder' module

More...

Asyncio support

PyUNICORE has a fully async implementation of the basic UNICORE APIs in the package pyunicore.aio.client. As an example, running a job would look like this:

import pyunicore.client as uc_client
import pyunicore.credentials as uc_credentials

base_url = "https://localhost:8080/DEMO-SITE/rest/core"
credential = uc_credentials.UsernamePassword("demouser", "test123")

async with uc_client.Client(credential, base_url) as client:
  my_job = {'Executable': 'date'}
  job = await client.new_job(job_description=my_job, inputs=[])
  await job.poll() # wait for job to finish
  work_dir = await job.working_dir
  stdout = await work_dir.stat("/stdout")
  content = await stdout.read()
  print(content)

More code examples can be found in the "integration-tests" folder in the source code repository.

Dask cluster implementation (experimental)

PyUNICORE provides an implementation of a Dask Cluster, allowing to run the Dask client on your local host (or in a Jupyter notebook in the Cloud), and have the Dask scheduler and workers running remotely on the HPC site.

More...

Helpers

The pyunicore.helpers module provides helper code for:

• Defining descriptions as a dataclass and easily converting to a dict as required by pyunicore.client.Client.new_job via a to_dict() method:
  • pyunicore.helpers.jobs.Description for pyunicore.client.Client.new_job()
  • pyunicore.helpers.workflows.Description for pyunicore.client.WorkflowService.new_workflow()
• Defining a workflow description

Defining a job or workflow

from pyunicore import helpers

client = ...

resources = helpers.jobs.Resources(nodes=4)
job = helpers.jobs.Description(
    executable="ls",
    project="demoproject",
    resources=resources
)

client.new_job(job.to_dict())

This works analogously for pyunicore.helpers.workflows.

Contributing

1. Fork the repository

2. Install the development dependencies

   pip install -r requirements-dev.txt

3. Install pre-commit hooks

   pre-commit install