cirro

 13class DataPortal:
 14    """
 15    Helper functions for exploring the Projects, Datasets, Samples, and Files
 16    available in the Data Portal.
 17    """
 18
 19    def __init__(self, base_url: str = None, client: CirroApi = None):
 20        """
 21        Set up the DataPortal object, establishing an authenticated connection.
 22
 23        Args:
 24            base_url (str): Optional base URL of the Cirro instance
 25             (if not provided, it uses the `CIRRO_BASE_URL` environment variable, or the config file)
 26            client (`cirro.cirro_client.CirroApi`): Optional pre-configured client
 27
 28        Example:
 29        ```python
 30        from cirro import DataPortal
 31
 32        Portal = DataPortal(base_url="app.cirro.bio")
 33        portal.list_projects()
 34        ```
 35        """
 36
 37        if client is not None:
 38            self._client = client
 39
 40        # Set up default client if not provided
 41        else:
 42            self._client = CirroApi(base_url=base_url)
 43
 44    def list_projects(self) -> DataPortalProjects:
 45        """List all the projects available in the Data Portal."""
 46
 47        return DataPortalProjects(
 48            [
 49                DataPortalProject(proj, self._client)
 50                for proj in self._client.projects.list()
 51            ]
 52        )
 53
 54    def get_project_by_name(self, name: str = None) -> DataPortalProject:
 55        """Return the project with the specified name."""
 56
 57        return self.list_projects().get_by_name(name)
 58
 59    def get_project_by_id(self, _id: str = None) -> DataPortalProject:
 60        """Return the project with the specified id."""
 61
 62        return self.list_projects().get_by_id(_id)
 63
 64    def get_project(self, project: str = None) -> DataPortalProject:
 65        """
 66        Return a project identified by ID or name.
 67
 68        Args:
 69            project (str): ID or name of project
 70
 71        Returns:
 72            `from cirro.sdk.project import DataPortalProject`
 73        """
 74        try:
 75            return self.get_project_by_id(project)
 76        except DataPortalAssetNotFound:
 77            return self.get_project_by_name(project)
 78
 79    def get_dataset(self, project: str = None, dataset: str = None) -> DataPortalDataset:
 80        """
 81        Return a dataset identified by ID or name.
 82
 83        Args:
 84            project (str): ID or name of project
 85            dataset (str): ID or name of dataset
 86
 87        Returns:
 88            `cirro.sdk.dataset.DataPortalDataset`
 89
 90            ```python
 91            from cirro import DataPortal()
 92            portal = DataPortal()
 93            dataset = portal.get_dataset(
 94                project="id-or-name-of-project",
 95                dataset="id-or-name-of-dataset"
 96            )
 97            ```
 98        """
 99        try:
100            project: DataPortalProject = self.get_project_by_id(project)
101        except DataPortalAssetNotFound:
102            project: DataPortalProject = self.get_project_by_name(project)
103
104        try:
105            return project.get_dataset_by_id(dataset)
106        except DataPortalAssetNotFound:
107            return project.get_dataset_by_name(dataset)
108
109    def list_processes(self, ingest=False) -> DataPortalProcesses:
110        """
111        List all the processes available in the Data Portal.
112        By default, only list non-ingest processes (those which can be run on existing datasets).
113        To list the processes which can be used to upload datasets, use `ingest = True`.
114
115        Args:
116            ingest (bool): If True, only list those processes which can be used to ingest datasets directly
117        """
118
119        return DataPortalProcesses(
120            [
121                DataPortalProcess(p, self._client)
122                for p in self._client.processes.list()
123                if not ingest or p.executor == Executor.INGEST
124            ]
125        )
126
127    def get_process_by_name(self, name: str, ingest=False) -> DataPortalProcess:
128        """
129        Return the process with the specified name.
130
131        Args:
132            name (str): Name of process
133        """
134
135        return self.list_processes(ingest=ingest).get_by_name(name)
136
137    def get_process_by_id(self, id: str, ingest=False) -> DataPortalProcess:
138        """
139        Return the process with the specified id
140
141        Args:
142            id (str): ID of process
143        """
144
145        return self.list_processes(ingest=ingest).get_by_id(id)
146
147    def list_reference_types(self) -> DataPortalReferenceTypes:
148        """
149        Return the list of all available reference types
150        """
151
152        return DataPortalReferenceTypes(
153            [
154                DataPortalReferenceType(ref)
155                for ref in self._client.references.get_types()
156            ]
157        )
158
159    @property
160    def developer_helper(self) -> DeveloperHelper:
161        return DeveloperHelper(self._client)

 8class DataPortalLogin:
 9    """
10    Start the login process, obtaining the authorization message from Cirro
11    needed to confirm the user identity.
12
13    Useful when you need to authenticate a user in a non-blocking way.
14
15    Usage:
16
17    ```python
18    # Replace app.cirro.bio as appropriate
19    login = DataPortalLogin(base_url="app.cirro.bio")
20
21    # Present the user with the authorization message
22    print(login.auth_message)
23
24    # Generate the authenticated DataPortal object,
25    # blocking until the user completes the login process in their browser
26    portal = login.await_completion()
27    ```
28    """
29    base_url: str
30    auth_info: DeviceCodeAuth
31
32    def __init__(self, base_url: str = None, enable_cache=False):
33        app_config = AppConfig(base_url=base_url)
34
35        self.base_url = base_url
36
37        self.auth_info = DeviceCodeAuth(
38            region=app_config.region,
39            client_id=app_config.client_id,
40            auth_endpoint=app_config.auth_endpoint,
41            enable_cache=enable_cache,
42            await_completion=False
43        )
44
45    @property
46    def auth_message(self) -> str:
47        """Authorization message provided by Cirro."""
48        return self.auth_info.auth_message
49
50    @property
51    def auth_message_markdown(self) -> str:
52        """Authorization message provided by Cirro (Markdown format)."""
53        return self.auth_info.auth_message_markdown
54
55    def await_completion(self) -> DataPortal:
56        """Complete the login process and return an authenticated client"""
57
58        # Block until the user completes the login flow
59        self.auth_info.await_completion()
60
61        # Set up the client object
62        cirro_client = CirroApi(
63            auth_info=self.auth_info,
64            base_url=self.base_url
65        )
66
67        # Return the Data Portal object
68        return DataPortal(client=cirro_client)

 20class DataPortalProject(DataPortalAsset):
 21    """
 22    Projects in the Data Portal contain collections of Datasets.
 23    Users are granted permissions at the project-level, allowing them
 24    to view and/or modify all the datasets in that collection.
 25    """
 26    def __init__(self, proj: Project, client: CirroApi):
 27        """
 28        Instantiate with helper method
 29
 30        ```python
 31        from cirro import DataPortal()
 32        portal = DataPortal()
 33        project = portal.get_project_by_name("Project Name")
 34        ```
 35
 36        """
 37        self._data = proj
 38        self._client = client
 39
 40    @property
 41    def id(self) -> str:
 42        """
 43        Unique identifier
 44        """
 45        return self._data.id
 46
 47    @property
 48    def name(self) -> str:
 49        """
 50        Readable name
 51        """
 52        return self._data.name
 53
 54    @property
 55    def description(self) -> str:
 56        """
 57        Longer description of the project
 58        """
 59        return self._data.description
 60
 61    @property
 62    def status(self) -> Status:
 63        """
 64        Status of the project
 65        """
 66        return self._data.status
 67
 68    def __str__(self):
 69        """Control how the Project is rendered as a string."""
 70
 71        return '\n'.join([
 72            f"{i.title()}: {self.__getattribute__(i)}"
 73            for i in ['name', 'id', 'description']
 74        ])
 75
 76    @cache
 77    def _get_datasets(self) -> List[Dataset]:
 78        return list_all_datasets(project_id=self.id,
 79                                 client=self._client)
 80
 81    def list_datasets(self, force_refresh=False) -> DataPortalDatasets:
 82        """List all the datasets available in the project."""
 83        if force_refresh:
 84            self._get_datasets.cache_clear()
 85
 86        return DataPortalDatasets(
 87            [
 88                DataPortalDataset(d, self._client)
 89                for d in self._get_datasets()
 90            ]
 91        )
 92
 93    def get_dataset_by_name(self, name: str, force_refresh=False) -> DataPortalDataset:
 94        """Return the dataset with the specified name."""
 95        if force_refresh:
 96            self._get_datasets.cache_clear()
 97
 98        dataset = next((d for d in self._get_datasets() if d.name == name), None)
 99        if dataset is None:
100            raise DataPortalAssetNotFound(f'Dataset with name {name} not found')
101        return self.get_dataset_by_id(dataset.id)
102
103    def get_dataset_by_id(self, _id: str = None) -> DataPortalDataset:
104        """Return the dataset with the specified id."""
105
106        dataset = self._client.datasets.get(project_id=self.id, dataset_id=_id)
107        if dataset is None:
108            raise DataPortalAssetNotFound(f'Dataset with ID {_id} not found')
109        return DataPortalDataset(dataset, self._client)
110
111    def list_references(self, reference_type: str = None) -> DataPortalReferences:
112        """
113        List the references available in a project.
114        Optionally filter to references of a particular type (identified by name)
115        """
116
117        # Get the complete list of references which are available
118        reference_types = DataPortalReferenceTypes(
119            [
120                DataPortalReferenceType(ref)
121                for ref in self._client.references.get_types()
122            ]
123        )
124
125        # If a particular name was specified
126        if reference_type is not None:
127            reference_types = reference_types.filter_by_pattern(reference_type)
128            if len(reference_types) == 0:
129                msg = f"Could not find any reference types with the name {reference_type}"
130                raise DataPortalAssetNotFound(msg)
131
132        return DataPortalReferences(
133            [
134                DataPortalReference(ref, project_id=self.id, client=self._client)
135                for ref in self._client.references.get_for_project(
136                    self.id
137                )
138                if reference_type is None or ref.type == reference_type
139            ]
140        )
141
142    def get_reference_by_name(self, name: str = None, ref_type: str = None) -> DataPortalReference:
143        """Return the reference of a particular type with the specified name."""
144
145        if name is None:
146            raise DataPortalInputError("Must specify the reference name")
147
148        return self.list_references(ref_type).get_by_name(name)
149
150    def upload_dataset(
151        self,
152        name: str = None,
153        description='',
154        process: Union[DataPortalProcess, str] = None,
155        upload_folder: str = None,
156        files: List[str] = None,
157        tags: List[str] = None,
158    ):
159        """
160        Upload a set of files to the Data Portal, creating a new dataset.
161
162        If the files parameter is not provided, it will upload all files in the upload folder
163
164        Args:
165            name (str): Name of newly created dataset
166            description (str): Description of newly created dataset
167            process (str | DataPortalProcess): Process to run may be referenced by name, ID, or object
168            upload_folder (str): Folder containing files to upload
169            files (List[str]): Optional subset of files to upload from the folder
170            tags (List[str]): Optional list of tags to apply to the dataset
171        """
172
173        if name is None:
174            raise DataPortalInputError("Must provide name for new dataset")
175        if process is None:
176            raise DataPortalInputError("Must provide the process which is used for ingest")
177        if upload_folder is None:
178            raise DataPortalInputError("Must provide upload_folder -- folder containing files to upload")
179
180        # Parse the process provided by the user
181        process = parse_process_name_or_id(process, self._client)
182
183        # If no files were provided
184        if files is None:
185            # Get the list of files in the upload folder
186            files = get_files_in_directory(upload_folder)
187
188        if files is None or len(files) == 0:
189            raise RuntimeWarning("No files to upload, exiting")
190
191        # Normalize into Tag object
192        if tags is not None:
193            tags = [Tag(value=value) for value in tags]
194
195        # Make sure that the files match the expected pattern
196        self._client.processes.check_dataset_files(files, process.id, upload_folder)
197
198        # Create the ingest process request
199        dataset_create_request = UploadDatasetRequest(
200            process_id=process.id,
201            name=name,
202            description=description,
203            expected_files=files,
204            tags=tags,
205        )
206
207        # Get the response
208        create_response = self._client.datasets.create(project_id=self.id,
209                                                       upload_request=dataset_create_request)
210
211        # Upload the files
212        self._client.datasets.upload_files(
213            project_id=self.id,
214            dataset_id=create_response.id,
215            directory=upload_folder,
216            files=files
217        )
218
219        # Return the dataset which was created, which might take a second to update
220        max_attempts = 5
221        for attempt in range(max_attempts):
222            try:
223                return self.get_dataset_by_id(create_response.id)
224            except DataPortalAssetNotFound as e:
225                if attempt == max_attempts - 1:
226                    raise e
227                else:
228                    sleep(2)
229
230    def samples(self, max_items: int = 10000) -> List[Sample]:
231        """
232        Retrieves a list of samples associated with a project along with their metadata
233
234        Args:
235            max_items (int): Maximum number of records to get (default 10,000)
236        """
237        return self._client.metadata.get_project_samples(self.id, max_items)

11class DataPortalProcess(DataPortalAsset):
12    """Helper functions for interacting with analysis processes."""
13
14    def __init__(self, process: Union[Process, ProcessDetail], 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 category(self) -> str:
54        """Category of process"""
55        return self._data.category
56
57    @property
58    def pipeline_type(self) -> str:
59        """Pipeline type"""
60        return self._data.pipeline_type
61
62    @property
63    def documentation_url(self) -> str:
64        """Documentation URL"""
65        return self._data.documentation_url
66
67    @property
68    def file_requirements_message(self) -> str:
69        """Description of files required for INGEST processes"""
70        return self._data.file_requirements_message
71
72    @property
73    def code(self) -> PipelineCode:
74        """Pipeline code configuration"""
75        return self._get_detail().pipeline_code
76
77    @property
78    def custom_settings(self) -> CustomPipelineSettings:
79        """Custom settings for the process"""
80        return self._get_detail().custom_settings
81
82    def _get_detail(self) -> ProcessDetail:
83        if not isinstance(self._data, ProcessDetail):
84            self._data = self._client.processes.get(self.id)
85        return self._data
86
87    def __str__(self):
88        return '\n'.join([
89            f"{i.title()}: {self.__getattribute__(i)}"
90            for i in ['name', 'id', 'description']
91        ])
92
93    def get_parameter_spec(self) -> ParameterSpecification:
94        """
95        Gets a specification used to describe the parameters used in the process.
96        """
97        return self._client.processes.get_parameter_spec(self.id)

 18class DataPortalDataset(DataPortalAsset):
 19    """
 20    Datasets in the Data Portal are collections of files which have
 21    either been uploaded directly, or which have been output by
 22    an analysis pipeline or notebook.
 23    """
 24
 25    def __init__(self, dataset: Union[Dataset, DatasetDetail], client: CirroApi):
 26        """
 27        Instantiate a dataset object
 28
 29        Should be invoked from a top-level constructor, for example:
 30
 31        ```python
 32        from cirro import DataPortal()
 33        portal = DataPortal()
 34        dataset = portal.get_dataset(
 35            project="id-or-name-of-project",
 36            dataset="id-or-name-of-dataset"
 37        )
 38        ```
 39
 40        """
 41        assert dataset.project_id is not None, "Must provide dataset with project_id attribute"
 42        self._data = dataset
 43        self._assets: Optional[DatasetAssets] = None
 44        self._client = client
 45
 46    @property
 47    def id(self) -> str:
 48        """Unique identifier for the dataset"""
 49        return self._data.id
 50
 51    @property
 52    def name(self) -> str:
 53        """Editable name for the dataset"""
 54        return self._data.name
 55
 56    @property
 57    def description(self) -> str:
 58        """Longer name for the dataset"""
 59        return self._data.description
 60
 61    @property
 62    def process_id(self) -> str:
 63        """Unique ID of process used to create the dataset"""
 64        return self._data.process_id
 65
 66    @property
 67    def process(self) -> ProcessDetail:
 68        """
 69        Object representing the process used to create the dataset
 70        """
 71        return self._client.processes.get(self.process_id)
 72
 73    @property
 74    def project_id(self) -> str:
 75        """ID of the project containing the dataset"""
 76        return self._data.project_id
 77
 78    @property
 79    def status(self) -> Status:
 80        """
 81        Status of the dataset
 82        """
 83        return self._data.status
 84
 85    @property
 86    def source_dataset_ids(self) -> List[str]:
 87        """IDs of the datasets used as sources for this dataset (if any)"""
 88        return self._data.source_dataset_ids
 89
 90    @property
 91    def source_datasets(self) -> List['DataPortalDataset']:
 92        """
 93        Objects representing the datasets used as sources for this dataset (if any)
 94        """
 95        return [
 96            DataPortalDataset(
 97                dataset=self._client.datasets.get(project_id=self.project_id, dataset_id=dataset_id),
 98                client=self._client
 99            )
100            for dataset_id in self.source_dataset_ids
101        ]
102
103    @property
104    def params(self) -> dict:
105        """
106        Parameters used to generate the dataset
107        """
108        return self._get_detail().params.to_dict()
109
110    @property
111    def info(self) -> dict:
112        """
113        Extra information about the dataset
114        """
115        return self._get_detail().info.to_dict()
116
117    @property
118    def tags(self) -> List[Tag]:
119        """
120        Tags applied to the dataset
121        """
122        return self._data.tags
123
124    @property
125    def share(self) -> Optional[NamedItem]:
126        """
127        Share associated with the dataset, if any.
128        """
129        return self._get_detail().share
130
131    @property
132    def created_by(self) -> str:
133        """User who created the dataset"""
134        return self._data.created_by
135
136    @property
137    def created_at(self) -> datetime.datetime:
138        """Timestamp of dataset creation"""
139        return self._data.created_at
140
141    def _get_detail(self):
142        if not isinstance(self._data, DatasetDetail):
143            self._data = self._client.datasets.get(project_id=self.project_id, dataset_id=self.id)
144        return self._data
145
146    def _get_assets(self):
147        if not self._assets:
148            self._assets = self._client.datasets.get_assets_listing(
149                project_id=self.project_id,
150                dataset_id=self.id
151            )
152        return self._assets
153
154    def __str__(self):
155        return '\n'.join([
156            f"{i.title()}: {self.__getattribute__(i)}"
157            for i in ['name', 'id', 'description', 'status']
158        ])
159
160    def get_file(self, relative_path: str) -> DataPortalFile:
161        """
162        Get a file from the dataset using its relative path.
163
164        Args:
165            relative_path (str): Relative path of file within the dataset
166
167        Returns:
168            `from cirro.sdk.file import DataPortalFile`
169        """
170
171        # Get the list of files in this dataset
172        files = self.list_files()
173
174        # Try getting the file using the relative path provided by the user
175        try:
176            return files.get_by_id(relative_path)
177        except DataPortalAssetNotFound:
178            # Try getting the file with the 'data/' prefix prepended
179            try:
180                return files.get_by_id("data/" + relative_path)
181            except DataPortalAssetNotFound:
182                # If not found, raise the exception using the string provided
183                # by the user, not the data/ prepended version (which may be
184                # confusing to the user)
185                msg = '\n'.join([f"No file found with path '{relative_path}'."])
186                raise DataPortalAssetNotFound(msg)
187
188    def list_files(self) -> DataPortalFiles:
189        """
190        Return the list of files which make up the dataset.
191        """
192        files = self._get_assets().files
193        return DataPortalFiles(
194            [
195                DataPortalFile(file=file, client=self._client)
196                for file in files
197            ]
198        )
199
200    def get_artifact(self, artifact_type: ArtifactType) -> DataPortalFile:
201        """
202        Get the artifact of a particular type from the dataset
203        """
204        artifacts = self._get_assets().artifacts
205        artifact = next((a for a in artifacts if a.artifact_type == artifact_type), None)
206        if artifact is None:
207            raise DataPortalAssetNotFound(f"No artifact found with type '{artifact_type}'")
208        return DataPortalFile(file=artifact.file, client=self._client)
209
210    def list_artifacts(self) -> List[DataPortalFile]:
211        """
212        Return the list of artifacts associated with the dataset
213
214        An artifact may be something generated as part of the analysis or other process.
215        See `cirro_api_client.v1.models.ArtifactType` for the list of possible artifact types.
216
217        """
218        artifacts = self._get_assets().artifacts
219        return DataPortalFiles(
220            [
221                DataPortalFile(file=artifact.file, client=self._client)
222                for artifact in artifacts
223            ]
224        )
225
226    def download_files(self, download_location: str = None) -> None:
227        """
228        Download all the files from the dataset to a local directory.
229
230        Args:
231            download_location (str): Path to local directory
232        """
233
234        # Alias for internal method
235        self.list_files().download(download_location)
236
237    def run_analysis(
238            self,
239            name: str = None,
240            description: str = "",
241            process: Union[DataPortalProcess, str] = None,
242            params=None,
243            notifications_emails: List[str] = None,
244            compute_environment: str = None,
245            resume_dataset_id: str = None
246    ) -> str:
247        """
248        Runs an analysis on a dataset, returns the ID of the newly created dataset.
249
250        The process can be provided as either a DataPortalProcess object,
251        or a string which corresponds to the name or ID of the process.
252
253        Args:
254            name (str): Name of newly created dataset
255            description (str): Description of newly created dataset
256            process (DataPortalProcess or str): Process to run
257            params (dict): Analysis parameters
258            notifications_emails (List[str]): Notification email address(es)
259            compute_environment (str): Name or ID of compute environment to use,
260             if blank it will run in AWS
261            resume_dataset_id (str): ID of dataset to resume from, used for caching task execution.
262             It will attempt to re-use the previous output to minimize duplicate work
263
264        Returns:
265            dataset_id (str): ID of newly created dataset
266        """
267        if name is None:
268            raise DataPortalInputError("Must specify 'name' for run_analysis")
269        if process is None:
270            raise DataPortalInputError("Must specify 'process' for run_analysis")
271        if notifications_emails is None:
272            notifications_emails = []
273        if params is None:
274            params = {}
275
276        # If the process is a string, try to parse it as a process name or ID
277        process = parse_process_name_or_id(process, self._client)
278
279        if compute_environment:
280            compute_environments = self._client.compute_environments.list_environments_for_project(
281                project_id=self.project_id
282            )
283            compute_environment = next(
284                (env for env in compute_environments
285                 if env.name == compute_environment or env.id == compute_environment),
286                None
287            )
288            if compute_environment is None:
289                raise DataPortalInputError(f"Compute environment '{compute_environment}' not found")
290
291        resp = self._client.execution.run_analysis(
292            project_id=self.project_id,
293            request=RunAnalysisRequest(
294                name=name,
295                description=description,
296                process_id=process.id,
297                source_dataset_ids=[self.id],
298                params=RunAnalysisRequestParams.from_dict(params),
299                notification_emails=notifications_emails,
300                resume_dataset_id=resume_dataset_id,
301                compute_environment_id=compute_environment.id if compute_environment else None
302            )
303        )
304        return resp.id