cloudos_cli.clos

This is the main class of the package.

Classes

Cloudos(cloudos_url, apikey, cromwell_token)

A simple class to contain the required connection information.
class cloudos_cli.clos.Cloudos(cloudos_url, apikey, cromwell_token)[source]

Bases: object

A simple class to contain the required connection information.

Parameters:

  • cloudos_url (string) – The CloudOS service url.

  • apikey (string) – Your CloudOS API key.

  • cromwell_token (string) – Cromwell server token. If None, apikey will be used instead.

abort_job(job, workspace_id, verify=True)[source]

Abort a job.

Parameters:

  • job (string) – The CloudOS job id of the job to abort.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

r – The server response

Return type:

requests.models.Response

apikey: str
cloudos_url: str
create_project(workspace_id, project_name, verify=True)[source]

Create a new project in CloudOS.

Parameters:

  • workspace_id (str) – The CloudOS workspace ID where the project will be created.

  • project_name (str) – The name for the new project.

  • verify ([bool | str], optional) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file. Default is True.

Returns:

The ID of the newly created project.

Return type:

str

Raises:

BadRequestException – If the request to create the project fails with a status code indicating an error.

cromwell_switch(workspace_id, action, verify=True)[source]

Restart Cromwell server.

Parameters:

  • workspace_id (string) – The CloudOS workspace id in which restart/stop Cromwell status.

  • action (string [restart|stop]) – The action to perform.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

r – The server response

Return type:

requests.models.Response

cromwell_token: str
detect_workflow(workflow_name, workspace_id, verify=True, last=False)[source]

Detects workflow type: nextflow or wdl.

Parameters:

  • workflow_name (string) – Name of the workflow.

  • workspace_id (string) – The CloudOS workspace id from to collect the workflows.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

wt – The workflow type detected

Return type:

string [‘nextflow’|’wdl’]

get_cromwell_status(workspace_id, verify=True)[source]

Get Cromwell server status from CloudOS.

Parameters:

  • workspace_id (string) – The CloudOS workspace id from to check the Cromwell status.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

r – The server response

Return type:

requests.models.Response

get_job_list(workspace_id, last_n_jobs=30, page=1, archived=False, verify=True, filter_status=None, filter_job_name=None, filter_project=None, filter_workflow=None, filter_job_id=None, filter_only_mine=False, filter_owner=None, filter_queue=None, last=False)[source]

Get jobs from a CloudOS workspace with optional filtering.

Fetches jobs page by page, applies all filters after fetching. Stops when enough jobs are collected or no more jobs are available.

Parameters:

  • workspace_id (string) – The CloudOS workspace id from to collect the jobs.

  • last_n_jobs ([int | 'all']) – How many of the last jobs from the user to retrieve. You can specify a very large int or ‘all’ to get all user’s jobs.

  • page (int) – Response page to get (ignored when using filters - starts from page 1).

  • archived (bool) – When True, only the archived jobs are retrieved.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

  • filter_status (string, optional) – Filter jobs by status (e.g., ‘completed’, ‘running’, ‘failed’).

  • filter_job_name (string, optional) – Filter jobs by name.

  • filter_project (string, optional) – Filter jobs by project name (will be resolved to project ID).

  • filter_workflow (string, optional) – Filter jobs by workflow name (will be resolved to workflow ID).

  • filter_job_id (string, optional) – Filter jobs by specific job ID.

  • filter_only_mine (bool, optional) – Filter to show only jobs belonging to the current user.

  • filter_owner (string, optional) – Filter jobs by owner username (will be resolved to user ID).

  • filter_queue (string, optional) – Filter jobs by queue name (will be resolved to queue ID). Only applies to jobs running in batch environment. Non-batch jobs are preserved in results as they don’t use queues.

  • last (bool, optional) – When workflows are duplicated, use the latest imported workflow (by date).

Returns:

r – A list of dicts, each corresponding to a jobs from the user and the workspace.

Return type:

list

get_job_logs(j_id, workspace_id, verify=True)[source]

Get the location of the logs for the specified job

get_job_results(j_id, workspace_id, verify=True)[source]

Get the location of the results for the specified job

get_job_status(j_id, verify=True)[source]

Get job status from CloudOS.

Parameters:

  • j_id (string) – The CloudOS job id of the job just launched.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

r – The server response

Return type:

requests.models.Response

get_job_workdir(j_id, workspace_id, verify=True)[source]

Get the working directory for the specified job

get_project_id_from_name(workspace_id, project_name, verify=True)[source]

Retrieve the project ID from its name.

Parameters:

  • workspace_id (str) – The CloudOS workspace ID to search for the project.

  • project_name (str) – The name of the project to search for.

  • verify ([bool | str], optional) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file. Default is True.

Returns:

The server response containing project details.

Return type:

dict

Raises:

BadRequestException – If the request to retrieve the project fails with a status code indicating an error.

get_project_list(workspace_id, verify=True, get_all=True, page=1, page_size=10, max_page_size=100)[source]

Get all the project from a CloudOS workspace.

Parameters:

  • workspace_id (string) – The CloudOS workspace id from to collect the projects.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

  • get_all (bool) – Whether to get all available curated workflows or just the indicated page.

  • page (int) – The page number to retrieve, from the paginated response.

  • page_size (int) – The number of workflows by page. From 1 to 1000.

  • max_page_size (int) – Max page size defined by the API server. It is currently 1000.

Returns:

r – The server response

Return type:

requests.models.Response

get_storage_contents(cloud_name, cloud_meta, container, path, workspace_id, verify)[source]

Retrieves the contents of a storage container from the specified cloud service.

This method fetches the contents of a specified path within a storage container on a cloud service (e.g., AWS S3 or Azure Blob). The request is authenticated using an API key and requires valid parameters such as the workspace ID and path.

Parameters:

  • cloud_name (str) – The name of the cloud service (e.g., ‘aws’ or ‘azure’).

  • container (str) – The name of the storage container or bucket.

  • path (str) – The file path or directory within the storage container.

  • workspace_id (str) – The identifier of the workspace or team.

  • verify (bool) – Whether to verify SSL certificates for the request.

Returns:

A list of contents retrieved from the specified cloud storage.

Return type:

list

Raises:

  • BadRequestException – If the request to retrieve the contents fails with a

  • status code indicating an error. –

get_user_info(verify=True)[source]

Gets user information from users/me endpoint

Parameters:

verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

r – The server response content

Return type:

requests.models.Response.content

get_workflow_content(workspace_id, workflow_name, verify=True, last=False, max_page_size=100)[source]

Retrieve the workflow content from API.

Parameters:

  • workspace_id (str) – The CloudOS workspace ID to search for the workflow.

  • workflow_name (str) – The name of the workflow to search for.

  • verify ([bool | str], optional) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file. Default is True.

Returns:

The server response containing workflow details.

Return type:

dict

Raises:

BadRequestException – If the request to retrieve the project fails with a status code indicating an error.

get_workflow_list(workspace_id, verify=True, get_all=True, page=1, page_size=10, max_page_size=100, archived_status=False)[source]

Get all the workflows from a CloudOS workspace.

Parameters:

  • workspace_id (string) – The CloudOS workspace id from to collect the workflows.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

  • get_all (bool) – Whether to get all available curated workflows or just the indicated page.

  • page (int) – The page number to retrieve, from the paginated response.

  • page_size (int) – The number of workflows by page. From 1 to 1000.

  • max_page_size (int) – Max page size defined by the API server. It is currently 1000.

  • archived_status (bool) – Whether to retrieve archived workflows or not.

Returns:

r – A list of dicts, each corresponding to a workflow.

Return type:

list

get_workflow_max_pagination(workspace_id, workflow_name, verify=True)[source]

Retrieve the workflows max pages from API.

Parameters:

  • workspace_id (str) – The CloudOS workspace ID to search for the workflow.

  • workflow_name (str) – The name of the workflow to search for.

  • verify ([bool | str], optional) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file. Default is True.

Returns:

The server response with max pagination for workflows.

Return type:

int

Raises:

BadRequestException – If the request to retrieve the project fails with a status code indicating an error.

is_module(workflow_name, workspace_id, verify=True, last=False)[source]

Detects whether the workflow is a system module or not.

System modules use fixed queues, so this check is important to properly manage queue selection.

Parameters:

  • workflow_name (string) – Name of the workflow.

  • workspace_id (string) – The CloudOS workspace id from to collect the workflows.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

True, if the workflow is a system module, false otherwise.

Return type:

bool

static process_job_list(r, all_fields=False)[source]

Process a job list from a self.get_job_list call.

Parameters:

  • r (list) – A list of dicts, each corresponding to a job from the user and the workspace.

  • all_fields (bool. Default=False) – Whether to return a reduced version of the DataFrame containing only the selected columns or the full DataFrame.

Returns:

df – A DataFrame with the requested columns from the jobs.

Return type:

pandas.DataFrame

static process_project_list(r, all_fields=False)[source]

Process a server response from a self.get_project_list call.

Parameters:

  • r (requests.models.Response) – A list of dicts, each corresponding to a project.

  • all_fields (bool. Default=False) – Whether to return a reduced version of the DataFrame containing only the selected columns or the full DataFrame.

Returns:

df – A DataFrame with the requested columns from the projects.

Return type:

pandas.DataFrame

static process_workflow_list(r, all_fields=False)[source]

Process a server response from a self.get_workflow_list call.

Parameters:

  • r (list) – A list of dicts, each corresponding to a workflow.

  • all_fields (bool. Default=False) – Whether to return a reduced version of the DataFrame containing only the selected columns or the full DataFrame.

Returns:

df – A DataFrame with the requested columns from the workflows.

Return type:

pandas.DataFrame

reorder_job_list(my_jobs_df, filename='my_jobs.csv')[source]

Save a job list DataFrame to a CSV file with renamed and ordered columns.

Parameters:

  • my_jobs_df (pandas.DataFrame) – A DataFrame containing job information from process_job_list.

  • filename (str) – The name of the file to save the DataFrame to. Default is ‘my_jobs.csv’.

Returns:

Saves the DataFrame to a CSV file with renamed and ordered columns.

Return type:

None

resolve_user_id(filter_owner, workspace_id, verify=True)[source]

Resolve a username or display name to a user ID.

Parameters:

  • filter_owner (str) – The username or display name to search for.

  • workspace_id (str) – The CloudOS workspace ID.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

The user ID corresponding to the filter_owner.

Return type:

str

Raises:

ValueError – If the user cannot be found or if there’s an error during the search.

save_job_list_to_csv(my_jobs_df, filename='my_jobs.csv')[source]
wait_job_completion(job_id, wait_time=3600, request_interval=30, verbose=False, verify=True)[source]

Checks job status from CloudOS and wait for its complation.

Parameters:

  • j_id (string) – The CloudOS job id of the job just launched.

  • wait_time (int) – Max time to wait (in seconds) to job completion.

  • request_interval (int) – Time interval (in seconds) to request job status.

  • verbose (bool) – Whether to output status on every request or not.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

A dict with three elements collected from the job status: ‘name’, ‘id’, ‘status’.

Return type:

dict

workflow_content_query(workspace_id, workflow_name, verify=True, query='workflowType', last=False)[source]
workflow_import(workspace_id, workflow_url, workflow_name, repository_project_id, workflow_docs_link='', repository_id=None, verify=True)[source]

Imports workflows to CloudOS.

Parameters:

  • workspace_id (string) – The CloudOS workspace id from to collect the projects.

  • workflow_url (string) – The URL of the workflow. Only Github or Bitbucket are allowed.

  • workflow_name (string) – A name for the imported pipeline in CloudOS.

  • repository_project_id (int) – The repository project ID.

  • workflow_docs_link (string) – Link to the documentation URL.

  • repository_id (int) – The repository ID. Only required for GitHub repositories.

  • verify ([bool|string]) – Whether to use SSL verification or not. Alternatively, if a string is passed, it will be interpreted as the path to the SSL certificate file.

Returns:

workflow_id – The newly imported worflow ID.

Return type:

string