Project Pipeline

Classes

class wrapica.project_pipelines.ICAv2AnalysisInput

Bases: object

Parent class for CWLAnalysisInput, NextflowAnalysisInput

create_analysis_input(): Implemented in subclass :return:

validate_input(): For structured inputs, one can first import the pipeline id and compare the input values against those specified in the pipeline, and compare required against non-required inputs For JSON based inputs, one will need to validate against the JSON schema. This is not implemented yet as waiting on PR https://github.com/common-workflow-language/cwl-utils/pull/288 We also want https://github.com/umccr-illumina/ica_v2/issues/162 resolved so we can validate against the top level workflow object. :return:

class wrapica.project_pipelines.ICAv2CWLEngineParameters(project_id=None, pipeline_id=None, analysis_output=None, logs_output=None, analysis_input=None, tags=None, analysis_storage_id=None, analysis_storage_size=None, cwltool_overrides=None)

Bases: ICAv2EngineParameters

The ICAv2 EngineParameters has the following properties

get_overrides_from_engine_parameters(inputs_overrides=None): Get the overrides from the engine parameters. Note overrides in the inputjson should NOT be overwritten :type inputs_overrides: Optional[Dict] :param inputs_overrides: :rtype: Dict :return:

workflow_language: Literal['CWL', 'NEXTFLOW'] = 'CWL'

class wrapica.project_pipelines.ICAv2CWLPipelineAnalysis(user_reference, project_id, pipeline_id, analysis_input, analysis_storage_id=None, analysis_storage_size=None, analysis_output_uri=None, ica_logs_uri=None, tags=None, cwltool_overrides=None)

Bases: ICAv2PipelineAnalysis

The ICAv2CWLPipelineAnalysis has the following properties

user_reference: str
input_json: Dict (cwl_inputs)
engine_parameters: ICAv2EngineParameters

The engineParameters then populate the following parameters for its parent class

pipeline_id: str, tags: ICAv2PipelineAnalysisTags, analysis_input: Union[CwlAnalysisJsonInput, NextflowAnalysisInput], analysis_output: str, analysis_storage_size: Optional[AnalysisStorageSize] = None

create_analysis()

Return type:: CreateCwlWithJsonInputAnalysis

launch_analysis(idempotency_key=None)

Return type:: AnalysisV4

set_engine_parameters(): Implemented in subclass :return:

class wrapica.project_pipelines.ICAv2CwlAnalysisJsonInput(input_json)

Bases: ICAv2AnalysisInput

Generate a CWL Analysis input from a JSON input

create_analysis_input(): Implemented in subclass :rtype: CwlAnalysisWithJsonInput :return:

deference_cwl_input_json()

set_input_json()

validate_input(): Cannot yet validate the input as we don’t have an easy way to collect a JSON schema for a CWL input :return:

class wrapica.project_pipelines.ICAv2EngineParameters(project_id=None, pipeline_id=None, analysis_input=None, analysis_output=None, logs_output=None, tags=None, analysis_storage_id=None, analysis_storage_size=None)

Bases: object

The ICAv2 EngineParameters has the following properties

check_launch_parameters(): Check the launch parameters have been set :return:

check_meta_parameters()

check_output_parameters(): Ensure that at analysis_output is set :return:

populate_placeholders_in_output_path(analysis_path): Populate placeholders in the output path :type analysis_path: Path :param analysis_path: :return:

set_launch_parameters(project_id=None, pipeline_id=None, analysis_storage_id=None, analysis_storage_size=None, analysis_input=None)

set_meta_parameters(tags=None)

set_output_parameters(analysis_output=None, logs_output=None): Set analysis output :type analysis_output: Optional[List[AnalysisOutputMapping]] :param analysis_output: :type logs_output: Optional[CreateAnalysisLogs] :param logs_output: :return:

update_engine_parameter(attribute_name, value)

workflow_language = None

class wrapica.project_pipelines.ICAv2NextflowAnalysisInput(input_json, cache_uri=None)

Bases: ICAv2AnalysisInput

Generate a Nextflow Analysis input from a JSON input - if the input is an icav2 uri, it is considered an analysis data input

create_analysis_input(): Implemented in subclass :rtype: NextflowAnalysisWithCustomInput :return:

deference_nextflow_input_json()

set_input_json()

validate_input(): Cannot yet validate the input as we dont have an easy way to collect a JSON schema for the Nextflow pipeline. We can do this for non-proprietary pipelines, but not for proprietary pipelines. User is expected to know the input format of the Nextflow pipeline they are using. :return:

class wrapica.project_pipelines.ICAv2NextflowEngineParameters(project_id=None, pipeline_id=None, analysis_output=None, logs_output=None, analysis_input=None, tags=None, analysis_storage_id=None, analysis_storage_size=None)

Bases: ICAv2EngineParameters

The ICAv2 EngineParameters has the following properties

workflow_language: Literal['CWL', 'NEXTFLOW'] = 'NEXTFLOW'

class wrapica.project_pipelines.ICAv2NextflowPipelineAnalysis(user_reference, project_id, pipeline_id, analysis_input, analysis_storage_id=None, analysis_storage_size=None, analysis_output_uri=None, ica_logs_uri=None, tags=None)

Bases: ICAv2PipelineAnalysis

The ICAv2CWLPipelineAnalysis has the following properties

user_reference: str
input_json: Dict (cwl_inputs)
engine_parameters: ICAv2EngineParameters

The engineParameters then populate the following parameters for its parent class

pipeline_id: str, tags: ICAv2PipelineAnalysisTags, analysis_input: Union[CwlAnalysisJsonInput, NextflowAnalysisInput], analysis_output: str, analysis_storage_size: Optional[AnalysisStorageSize] = None

create_analysis()

Return type:: CreateNextflowWithCustomInputAnalysis

launch_analysis(idempotency_key=None)

Return type:: AnalysisV4

set_engine_parameters(): Implemented in subclass :return:

class wrapica.project_pipelines.ICAv2PipelineAnalysisTags(technical_tags, user_tags, reference_tags)

Bases: object

List of tags

clean_tags()

combine_tags(analysis_tags): Combine tags from another tag object :type analysis_tags: ICAv2PipelineAnalysisTags :param analysis_tags: :rtype: ICAv2PipelineAnalysisTags :return:

classmethod from_dict(tags_dict)

Functions

wrapica.project_pipelines.add_pipeline_file(project_id, pipeline_id, file_path, relative_path=None)

Add a pipeline file to a pipeline on icav2

Parameters:

project_id (Union[UUID, str]) – The project id to add the file to
pipeline_id (Union[UUID, str]) – The pipeline id to add the file to
file_path (Path) – The file path to add to the pipeline
relative_path (Optional[Path])

Returns:

The pipeline file object

Return type:

PipelineFile

Raises:

ApiException

Examples:

# Imports
from wrapica.project_pipelines import add_pipeline_file

# Add pipeline file
project_id = "project-123"
pipeline_id = "pipeline-123"
file_path = Path("/path/to/file.txt")
pipeline_file = add_pipeline_file(project_id, pipeline_id, file_path)

wrapica.project_pipelines.coerce_analysis_storage_id_or_size_to_analysis_storage(project_id, analysis_storage_id_or_size)

Given either an analysis storage id or analysis storage size, return the analysis storage id

Parameters:

project_id (Union[UUID, str])
analysis_storage_id_or_size (Union[str, Literal['Small', 'Medium', 'Large', 'XLarge', '2XLarge', '3XLarge']])

Return type:

Union[AnalysisStorageV3, AnalysisStorageV4]

Returns:

wrapica.project_pipelines.coerce_pipeline_id_or_code_to_project_pipeline_obj(pipeline_id_or_code)

Coerce a pipeline id or code to a project pipeline object

Parameters:: pipeline_id_or_code (str)
Return type:: Union[ProjectPipeline, ProjectPipelineV4]
Returns:: The project pipeline object
Raises:: ValueError, ApiException
Examples:

from wrapica.project_pipelines import coerce_pipeline_id_or_code_to_project_pipeline_obj

pipeline_id_or_code = "pipeline-123"

project_pipeline_obj = coerce_pipeline_id_or_code_to_project_pipeline_obj(pipeline_id_or_code)

wrapica.project_pipelines.convert_icav2_uris_to_data_ids_from_cwl_input_json(input_obj)

Return type:: Tuple[Union[str, Dict, List], List[AnalysisInputDataMount], List[AnalysisInputExternalData]]

wrapica.project_pipelines.convert_uris_to_data_ids_from_cwl_input_json(input_obj)

From a cwl input json, convert all the icav2 uris to data ids

Parameters:: input_obj (Union[str, int, float, bool, Dict, List]) – The CWL input object to convert
Returns:: The converted input object, mount list and external data list
Return type:: Tuple[Union[str, int, float, bool, Dict, List], List[AnalysisInputDataMount], List[AnalysisInputExternalData]]
Raises:: ValueError, ApiException
Examples:

from wrapica.project_pipelines import convert_icav2_uris_to_data_ids_from_cwl_input_json

input_obj = {
    "input_file": {
        "class": "File",
        "location": "icav2://project-123/data-path/file.txt"
    }
}

input_obj_new, mount_list, external_data_list = convert_icav2_uris_to_data_ids_from_cwl_input_json(
    input_obj
)

print(input_obj_new)
# Output: {
#   "input_file": {
#     "class": "File",
#     "location": "path/to/mount/file.txt"
#   }
# }

print(mount_list)
# Output: [
#   AnalysisInputDataMount(
#     dataId="fil.1234567890",
#     mountPath="path/to/mount/file.txt"
#   )

print(external_data_list)
# Output: []

wrapica.project_pipelines.create_blank_params_xml(output_file_path)

Create a params.xml file with no inputs

Parameters:: output_file_path (Path) – The output file path we wish to write the file to
Returns:: None
Examples:

# Imports
from wrapica.project_pipelines import create_blank_params_xml

# Create blank params file
output_file_path = Path("/path/to/params.xml")
create_blank_params_xml(output_file_path)

wrapica.project_pipelines.create_cwl_input_json_analysis_obj(user_reference, project_id, pipeline_id, analysis_input_dict, analysis_storage_id=None, analysis_storage_size=None, analysis_output_uri=None, tags=None, cwltool_overrides=None)

Given a pipeline id (optional - can be in the ICAv2EngineParameters An input json where the location attributes point to icav2 uris Generate a CreateCwlAnalysisWithJsonInput object ready for launch

Parameters:

user_reference (str) – The user reference to use for the analysis
project_id (Union[UUID, str]) – The project id that the pipeline exists in
pipeline_id (Union[UUID, str]) – The pipeline id to launch
analysis_input_dict (Dict) – The analysis input dictionary
analysis_storage_id (Union[UUID, str, None]) – The analysis storage id to use
analysis_storage_size (Optional[Literal['Small', 'Medium', 'Large', 'XLarge', '2XLarge', '3XLarge']]) – The analysis storage size to use
analysis_output_uri (Optional[str]) – The analysis output uri to use
tags (Optional[ICAv2PipelineAnalysisTags]) – The tags to use
cwltool_overrides (Optional[Dict]) – The cwltool overrides to use

Returns:

The CWL analysis object

Return type:

ICAv2CWLPipelineAnalysis

Examples:

from wrapica.project_analysis import (
    # Classes
    ICAv2PipelineAnalysisTags,
)

from wrapica.project_pipelines import (
    # Functions
    create_cwl_input_json_analysis_obj,
)

user_reference = "user-123"
project_id = "project-123"
pipeline_id = "pipeline-123"
analysis_input_dict = {
    "my_input_parameter": {
      "class": "File",
      "location": "icav2://project-123/data-path/file.txt"
    }
}
analysis_storage_id = "analysis-storage-123"
analysis_storage_size = AnalysisStorageSize.SMALL
analysis_output_uri = "icav2://project-123/output-path"
tags = ICAv2PipelineAnalysisTags(
    technical_tags=[
      "my_technical_tag",
    ]
    user_tags=[
      "user='John'",
      "billing='ExpensiveGroup'"
    ]
)

cwl_pipeline_analysis = create_cwl_input_json_analysis_obj(
    user_reference=user_reference,
    project_id=project_id,
    pipeline_id=pipeline_id,
    analysis_input_dict=analysis_input_dict,
    analysis_storage_id=analysis_storage_id,
    analysis_storage_size=analysis_storage_size,
    analysis_output_uri=analysis_output_uri,
    tags=tags
)

wrapica.project_pipelines.create_cwl_project_pipeline(project_id, pipeline_code, workflow_path, tool_paths=None, workflow_description=None, params_xml_file=None, analysis_storage=None, workflow_html_documentation=None, resource_type=None)

Create a CWL project pipeline from a workflow path and tool paths

Parameters:

resource_type (Optional[Literal['f1', 'f2', 'gpu', 'software_only']])
project_id (Union[UUID, str])
pipeline_code (str)
workflow_path (Path)
tool_paths (Optional[List[Path]])
workflow_description (Optional[str])
params_xml_file (Optional[Path])
analysis_storage (Union[AnalysisStorageV3, AnalysisStorageV4, None])
workflow_html_documentation (Optional[Path])

Return type:

ProjectPipelineV4

Returns:

wrapica.project_pipelines.create_cwl_workflow_from_zip(project_id, pipeline_code, zip_path, analysis_storage=None, workflow_description=None, html_documentation_path=None, resource_type=None)

Create a CWL project pipeline from a zip file containing the workflow and tools

Parameters:

project_id (Union[UUID, str])
pipeline_code (str)
zip_path (Path)
analysis_storage (Union[AnalysisStorageV3, AnalysisStorageV4, None])
workflow_description (Optional[str])
html_documentation_path (Optional[Path])
resource_type (Optional[Literal['f1', 'f2', 'gpu', 'software_only']])

Return type:

ProjectPipelineV4

Returns:

wrapica.project_pipelines.create_nextflow_pipeline_from_nf_core_zip(project_id, pipeline_code, zip_path, pipeline_revision, workflow_description=None, html_documentation_path=None, resource_type=None)

Create a Nextflow project pipeline from a zip file containing the workflow and tools This function is designed for a user to generate an nf-core pipeline from a zip file containing the workflow and tools.

Parameters:

pipeline_revision (str)
project_id (Union[UUID, str])
pipeline_code (str)
zip_path (Path)
workflow_description (Optional[str])
html_documentation_path (Optional[Path])
resource_type (Optional[Literal['f1', 'f2', 'gpu', 'software_only']])

Returns:

The nextflow pipeline

Return type:

ProjectPipeline

Examples:

:linenos:

# Bash Prestep
# Create a nextflow pipeline from a zip file containing the workflow and tools
# nf-core download bamtofastq --compress zip --outdir bamtofastq

# Imports
from wrapica.project_pipelines import create_nextflow_pipeline_from_zip

# Set vars
project_id = "project-123"
pipeline_code = "pipeline-123"
zip_path = Path("/path/to/bamtofastq.zip")
workflow_description = "This is the nf-core pipeline for bamtofastq"

# Create nextflow pipeline from zip
nextflow_pipeline = create_nextflow_pipeline_from_zip(
    project_id, pipeline_code, zip_path, workflow_description
)

wrapica.project_pipelines.create_nextflow_pipeline_from_zip(project_id, pipeline_code, zip_path, workflow_description, html_documentation_path=None, resource_type=None)

Return type:: ProjectPipelineV4

wrapica.project_pipelines.create_nextflow_project_pipeline(project_id, pipeline_code, main_nextflow_file, nextflow_config_file, other_nextflow_files, workflow_description, params_xml_file=None, analysis_storage=None, workflow_html_documentation=None, resource_type=None)

Create a CWL project pipeline from a workflow path and tool paths

Parameters:

project_id (Union[UUID, str])
pipeline_code (str)
main_nextflow_file (Path)
nextflow_config_file (Path)
other_nextflow_files (List[Path])
workflow_description (str)
params_xml_file (Optional[Path])
analysis_storage (Union[AnalysisStorageV3, AnalysisStorageV4, None])
workflow_html_documentation (Optional[Path])
resource_type (Optional[Literal['f1', 'f2', 'gpu', 'software_only']])

Return type:

ProjectPipelineV4

Returns:

wrapica.project_pipelines.create_params_xml(inputs, output_path)

From the inputs, create a params xml file

Parameters:

inputs (List[Union[WorkflowInputParameter, WorkflowInputParameter]])
output_path (Path)

Returns:

Examples:

# Imports
from wrapica.project_pipelines import create_params_xml

# Create params xml file
output_path = Path("/path/to/params.xml")
create_params_xml([], output_path)

wrapica.project_pipelines.delete_pipeline_file(project_id, pipeline_id, file_id)

Delete the pipeline file on icav2

Parameters:

project_id (Union[UUID, str])
pipeline_id (Union[UUID, str])
file_id (Union[UUID, str])

Raises:

ApiException

Examples:

# Imports
from wrapica.project_pipelines import update_pipeline_file

# Set vars
project_id = "project-123"
pipeline_id = "pipeline-123"

# Find pipefile where the file name is 'tabix-tool.cwl'
# And delete it
file_id = next(
    filter(
        lambda file_iter: file_iter.name == "tabix-tool.cwl",
        get_pipeline_files(project_id, pipeline_id)
    )
).id

delete_pipeline_file(project_id, pipeline_id, file_id)

wrapica.project_pipelines.get_analysis_storage_from_analysis_storage_id(project_id, analysis_storage_id)

Given an analysis storage id, return the analysis storage object :type project_id: Union[UUID, str] :param project_id: :type analysis_storage_id: Union[UUID, str] :param analysis_storage_id:

Returns:: The analysis storage object
Return type:: AnalysisStorage
Raises:: ValueError, ApiException

wrapica.project_pipelines.get_analysis_storage_from_analysis_storage_size(project_id, analysis_storage_size)

Given an analysis storage size, return the analysis storage object

Parameters:

project_id (Union[UUID, str])
analysis_storage_size (Literal['Small', 'Medium', 'Large', 'XLarge', '2XLarge', '3XLarge'])

Return type:

Union[AnalysisStorageV3, AnalysisStorageV4]

Returns:

wrapica.project_pipelines.get_analysis_storage_id_from_analysis_storage_size(project_id, analysis_storage_size)

Given an analysis storage size, return the analysis storage id

Parameters:

project_id (Union[UUID, str]) – The project id that the analysis storage exists in
analysis_storage_size (Literal['Small', 'Medium', 'Large', 'XLarge', '2XLarge', '3XLarge']) – The analysis storage size to retrieve the id for

Returns:

The analysis storage id

Return type:

str

Raises:

ValueError, ApiException

Examples:

from wrapica.project_pipelines import get_analysis_storage_id_from_analysis_storage_size
from wrapica.enums import AnalysisStorageSize

analysis_storage_size = AnalysisStorageSize.SMALL

analysis_storage_id = get_analysis_storage_id_from_analysis_storage_size(analysis_storage_size)

wrapica.project_pipelines.get_default_analysis_storage_id_from_project_pipeline(project_id, pipeline_id)

Given a project id and pipeline id, return the default analysis storage id for that pipeline

Parameters:

project_id (Union[UUID, str]) – The project id that the pipeline exists in
pipeline_id (Union[UUID, str]) – The pipeline id to retrieve the analysis storage information from

Returns:

The analysis storage id

Return type:

str

Raises:

ValueError, ApiException

Examples:

from wrapica.project_pipelines import get_default_analysis_storage_id_from_project_pipeline

project_id = "project-123"
pipeline_id = "pipeline-123"

# Use get_project_pipeline_id_from_pipeline_code to get the pipeline id

analysis_storage_id = get_default_analysis_storage_id_from_project_pipeline(project_id, pipeline_id)

wrapica.project_pipelines.get_default_analysis_storage_obj_from_project_pipeline(project_id, pipeline_id)

Given a project id and pipeline id, return the default analysis storage object for that pipeline

Parameters:

project_id (Union[UUID, str])
pipeline_id (Union[UUID, str])

Returns:

The analysis storage

Return type:

AnalysisStorage

Raises:

ValueError, ApiException

Examples:

from wrapica.project_pipelines import get_default_analysis_storage_id_from_project_pipeline

project_id = "project-123"
pipeline_id = "pipeline-123"

# Use get_project_pipeline_id_from_pipeline_code to get the pipeline id

analysis_storage_obj = get_default_analysis_storage_obj_from_project_pipeline(project_id, pipeline_id)

wrapica.project_pipelines.get_project_pipeline_configuration_parameters(project_id, pipeline_id)

Given a pipeline and project id, return the configuration parameters for the pipeline

Parameters:

project_id (Union[UUID, str])
pipeline_id (Union[UUID, str])

Returns:

The configuration parameters for the project pipeline

Return type:

List[PipelineConfigurationParameter]

Raises:

ApiException

Examples:

from wrapica.project_pipelines import (
    get_project_pipeline_configuration_parameters
)
from wrapica.libica_models import PipelineConfigurationParameter

project_id = "project-123"
pipeline_id = "pipeline-123"

configuration_parameters: List[PipelineConfigurationParameter] = (
    get_project_pipeline_configuration_parameters(project_id, pipeline_id)
)

for configuration_parameter in configuration_parameters:
    print(configuration_parameter.code)
    print(configuration_parameter.required)
    print(configuration_parameter.multi_value)
    print(configuration_parameter.type)

# Output:
# configuration_parameter_1
# false
# true
# boolean

wrapica.project_pipelines.get_project_pipeline_description_from_pipeline_id(project_id, pipeline_id)

Get a project pipeline description from a project id and pipeline id

Parameters:

project_id (Union[UUID, str]) – The project id that the pipeline exists in
pipeline_id (Union[UUID, str]) – The pipeline id to retrieve the description from

Returns:

The pipeline description

Return type:

str

Raises:

ValueError, ApiException

Examples:

from wrapica.project_pipelines import get_project_pipeline_description_from_pipeline_id

project_id = "project-123"
pipeline_id = "pipeline-123"

# Use get_project_pipeline_id_from_pipeline_code to get the pipeline id
pipeline_description = get_project_pipeline_description_from_pipeline_id(project_id, pipeline_id)

wrapica.project_pipelines.get_project_pipeline_id_from_pipeline_code(project_id, pipeline_code)

Given a project pipeline code and project id, return the pipeline id

Parameters:

project_id (Union[UUID, str]) – The project id that the pipeline exists in
pipeline_code (str) – The pipeline code to retrieve

Returns:

The pipeline id

Return type:

str

Raises:

ValueError: If the pipeline cannot be found

Examples:

from wrapica.project_pipelines import get_project_pipeline_id_from_pipeline_code

project_id = "project-123"
pipeline_code = "pipeline-123"

pipeline_id = get_project_pipeline_id_from_pipeline_code(project_id, pipeline_code)

wrapica.project_pipelines.get_project_pipeline_input_parameters(project_id, pipeline_id)

Get project pipeline input parameters, needed for structured input validation

Parameters:

project_id (Union[UUID, str])
pipeline_id (Union[UUID, str])

Returns:

The input parameters for the project pipeline

Return type:

List[InputParameter]

Raises:

ApiException

Examples:

from wrapica.project_pipelines import (
    # Functions
    get_project_pipeline_input_parameters
)
from wrapica.libica_models import InputParameter

project_id = "project-123"
pipeline_id = "pipeline-123"

input_parameters: List[InputParameter] = get_project_pipeline_input_parameters(project_id, pipeline_id)

for input_parameter in input_parameters:
    print(input_parameter.code)
    print(input_parameter.required)
    print(input_parameter.multi_value)

# Output:
# input_parameter_1
# false
# true

wrapica.project_pipelines.get_project_pipeline_obj(project_id, pipeline_id)

Given a project id and pipeline id, return the project pipeline object

Parameters:

project_id (Union[UUID, str]) – The project id that the pipeline exists in
pipeline_id (Union[UUID, str]) – The pipeline id to retrieve

Returns:

The project pipeline object

Return type:

ProjectPipeline

Raises:

ValueError: If the pipeline cannot be found

Examples:

from wrapica.project_pipelines import get_project_pipeline_obj

project_id = "project-123"
pipeline_id = "pipeline-123"

project_pipeline_obj = get_project_pipeline_obj(project_id, pipeline_id)

wrapica.project_pipelines.get_project_pipeline_obj_from_pipeline_code(project_id, pipeline_code)

Given a project id and pipeline code, return the project pipeline object

Parameters:

project_id (Union[UUID, str]) – The project id that the pipeline exists in
pipeline_code (str) – The pipeline code to retrieve

Returns:

The pipeline id

Return type:

str

Raises:

ValueError: If the pipeline cannot be found

Examples:

from wrapica.project_pipelines import get_project_pipeline_id_from_pipeline_code

project_id = "project-123"
pipeline_code = "pipeline-123"

pipeline_id = get_project_pipeline_id_from_pipeline_code(project_id, pipeline_code)

wrapica.project_pipelines.is_pipeline_in_project(project_id, pipeline_id)

Check if a pipeline is in a project

Parameters:

project_id (Union[UUID, str]) – The project id to check
pipeline_id (Union[UUID, str]) – The pipeline id to check

Returns:

True if the pipeline is in the project, False otherwise

Return type:

bool

Examples:

# Imports
from wrapica.project_pipelines import is_pipeline_in_project

# Check if pipeline is in project
project_id = "project-123"
pipeline_id = "pipeline-123"
pipeline_is_in_project = is_pipeline_in_project(project_id, pipeline_id)

wrapica.project_pipelines.launch_cwl_workflow(project_id, cwl_analysis, idempotency_key=None)

Launch a CWL Workflow in a specific project context

Parameters:

project_id (Union[UUID, str]) – The project id to launch the CWL workflow in
cwl_analysis (CreateCwlWithJsonInputAnalysis) – The CWL analysis object to launch
idempotency_key – The Idempotency-Key header can be used to prevent duplicate requests and support retries.

Returns:

the analysis ID along with the deconstructed json used for submission to the end point

Return type:

Analysis

Examples:

from pathlib import Path
from wrapica.project_pipelines import (
    # Functions
    launch_cwl_workflow,
    # Wrapica classes
    ICAv2CWLPipelineAnalysis,
)

from wrapica.libica_models import CreateCwlWithJsonInputAnalysis

# Initialise an ICAv2CWLPipeline Analysis object
cwl_analysis = ICAv2CWLPipelineAnalysis(
    user_reference="user-123",
    project_id="project-123",
    pipeline_id="pipeline-123",
    analysis_input={
        "input": "json"
    }
)

# Generate the inputs and analysis object
cwl_analysis.check_engine_parameters()
cwl_analysis.create_analysis()

# Launch the analysis pipeline
analysis = launch_cwl_workflow(project_id, cwl_analysis.analysis)

# Alternatively, just call cwl_analysis and it will launch the pipeline.
# analysis = cwl_analysis()

# Save the analysis
cwl_analysis.save_analysis(Path("/path/to/analysis.json"))

wrapica.project_pipelines.launch_nextflow_workflow(project_id, nextflow_analysis, idempotency_key=None)

Launch a Nextflow Workflow in a specific project context

Parameters:

project_id (Union[UUID, str]) – The project id to launch the Nextflow workflow in
nextflow_analysis (CreateNextflowWithCustomInputAnalysis) – The Nextflow analysis object to launch
idempotency_key – Prevent duplicate requests and support retries by providing an Idempotency-Key header.

Returns:

the analysis ID along with the deconstructed json used for submission to the end point

Return type:

Analysis

Examples:

from pathlib import Path
from wrapica.project_pipelines import (
    # Functions
    launch_nextflow_workflow,
    # Wrapica classes
    ICAv2NextflowPipelineAnalysis,
)

from wrapica.libica_models import CreateNextflowAnalysis, Analysis

# Initialise an ICAv2CWLPipeline Analysis object
nextflow_analysis = ICAv2NextflowPipelineAnalysis(
    user_reference="user-123",
    project_id="project-123",
    pipeline_id="pipeline-123",
    analysis_input={
        "my_input_parameter": "icav2://path/to/data",
        "my_config_parameter": "value"
    }
)

# Generate the inputs and analysis object
nextflow_analysis.check_engine_parameters()
nextflow_analysis.create_analysis()

# Launch the analysis pipeline
analysis = launch_nextflow_workflow(project_id, nextflow_analysis.analysis)

# Alternatively, just call cwl_analysis and it will launch the pipeline.
# analysis = nextflow_analysis()

# Save the analysis
nextflow_analysis.save_analysis(Path("/path/to/analysis.json"))

wrapica.project_pipelines.list_project_pipelines(project_id)

List pipelines in project

Parameters:

project_id (Union[UUID, str]) – List all pipelines avialable to this project

Returns:

The list of pipelines

Return type:

List[ProjectPipeline]

Raises:

ValueError, ApiException

Examples:

# Imports
import json
from wrapica.project_pipelines import list_pipelines_in_project

# Get list of pipelines in project
project_id = "project-123"
pipeline_list = list_pipelines_in_project(project_id)

print(
  json.dumps(
    map(
        lambda: pipeline_iter: {
            "id": pipeline_iter.id,
            "code": pipeline_iter.code,
        },
        pipeline_list
    )
  )
)

wrapica.project_pipelines.list_projects_with_pipeline(pipeline_id, include_hidden_projects)

Given a pipeline id, return a list of projects that the pipeline is linked to

Parameters:

pipeline_id (Union[UUID, str]) – The pipeline id to check
include_hidden_projects (bool) – Include hidden projects in the list

Returns:

The list of projects

Return type:

rtype:: List[Project]

Raises:

ValueError, ApiException

Examples:

from wrapica.project_pipelines import list_projects_with_pipeline

pipeline_id = "pipeline-123"
project_id = next(list_projects_with_pipeline(pipeline_id, include_hidden_projects=False)).id

project_pipeline_obj = get_project_pipeline_obj(project_id, pipeline_id)

wrapica.project_pipelines.release_project_pipeline(project_id, pipeline_id)

Convert a project pipeline from a draft status to a released status

Parameters:

project_id (Union[UUID, str])
pipeline_id (Union[UUID, str])

Raises:

ValueError – If the pipeline is not in draft status, or if the pipeline does not belong to the user
ApiException – If the API call fails

wrapica.project_pipelines.update_pipeline_file(project_id, pipeline_id, file_id, file_path)

Update the pipeline file on icav2

Parameters:

project_id (Union[UUID, str])
pipeline_id (Union[UUID, str])
file_id (Union[UUID, str])
file_path (Path)

Raises:

ApiException

Examples:

# Imports
from wrapica.project_pipelines import update_pipeline_file

# Set vars
project_id = "project-123"
pipeline_id = "pipeline-123"

# Find pipefile where the file name is 'tabix-tool.cwl'
file_id = next(
    filter(
        lambda file_iter: file_iter.name == "tabix-tool.cwl",
        get_pipeline_files(project_id, pipeline_id)
    )
).id

tool_file_with_new_content = Path("/path/to/tabix-tool.cwl")

update_pipeline_file(project_id, pipeline_id, file_id, file_path_with_new_content)