cirro.sdk.process

 1from typing import List
 2
 3from cirro_api_client.v1.models import Process, Executor
 4
 5from cirro.cirro_client import CirroApi
 6from cirro.models.form_specification import ParameterSpecification
 7from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
 8
 9
10class DataPortalProcess(DataPortalAsset):
11    """Helper functions for interacting with analysis processes."""
12    _data: Process
13
14    def __init__(self, process: Process, client: CirroApi):
15        """
16        Instantiate with helper method
17
18        ```python
19        from cirro import DataPortal()
20        portal = DataPortal()
21        process = portal.get_process_by_name("Process Name")
22        ```
23        """
24        self._data = process
25        self._client = client
26
27    @property
28    def id(self) -> str:
29        """Unique identifier"""
30        return self._data.id
31
32    @property
33    def name(self) -> str:
34        """Readable name"""
35        return self._data.name
36
37    @property
38    def description(self) -> str:
39        """Longer description of process"""
40        return self._data.description
41
42    @property
43    def child_process_ids(self) -> List[str]:
44        """List of processes which can be run on the output of this process"""
45        return self._data.child_process_ids
46
47    @property
48    def executor(self) -> Executor:
49        """INGEST, CROMWELL, or NEXTFLOW"""
50        return self._data.executor
51
52    @property
53    def documentation_url(self) -> str:
54        """Documentation URL"""
55        return self._data.documentation_url
56
57    @property
58    def file_requirements_message(self) -> str:
59        """Description of files required for INGEST processes"""
60        return self._data.file_requirements_message
61
62    def __str__(self):
63        return '\n'.join([
64            f"{i.title()}: {self.__getattribute__(i)}"
65            for i in ['name', 'id', 'description']
66        ])
67
68    def get_parameter_spec(self) -> ParameterSpecification:
69        """
70        Gets a specification used to describe the parameters used in the process.
71        """
72        return self._client.processes.get_parameter_spec(self.id)
73
74
75class DataPortalProcesses(DataPortalAssets[DataPortalProcess]):
76    """Collection of DataPortalProcess objects."""
77    asset_name = "process"