cirro.services

10class BillingService(BaseService):
11    """
12    Service for interacting with the Billing endpoints
13    """
14    def list(self) -> List[BillingAccount]:
15        """
16        Gets a list of billing accounts the current user has access to
17        """
18        return get_billing_accounts.sync(client=self._api_client)
19
20    def update(self, billing_account_id: str, request: BillingAccountRequest):
21        """
22        Updates a billing account
23
24        Args:
25            billing_account_id (str): Billing account ID
26            request (cirro_api_client.v1.models.BillingAccountRequest):
27
28        ```python
29        from cirro_api_client.v1.models import BillingAccountRequest
30        from cirro.cirro_client import CirroApi
31
32        cirro = CirroApi()
33        request = BillingAccountRequest(
34            name="New billing account name",
35            primary_budget_number="new-budget-number",
36            owner="New Owner"
37        )
38        cirro.billing.update("billing-account-id", request)
39        ```
40        """
41        update_billing_account.sync_detailed(
42            billing_account_id=billing_account_id,
43            body=request,
44            client=self._api_client
45        )

 21class DatasetService(FileEnabledService):
 22    """
 23    Service for interacting with the Dataset endpoints
 24    """
 25
 26    def list(self, project_id: str, max_items: int = 10000) -> List[Dataset]:
 27        """List datasets
 28
 29        Retrieves a list of datasets for a given project
 30
 31        Args:
 32            project_id (str): ID of the Project
 33            max_items (int): Maximum number of records to get (default 10,000)
 34        """
 35        return get_all_records(
 36            records_getter=lambda page_args: get_datasets.sync(
 37                project_id=project_id,
 38                client=self._api_client,
 39                next_token=page_args.next_token,
 40                limit=page_args.limit
 41            ),
 42            max_items=max_items
 43        )
 44
 45    def list_shared(self, project_id: str, share_id: str, max_items: int = 10000) -> List[Dataset]:
 46        """
 47        Retrieves a list of shared datasets for a given project and share
 48
 49        Args:
 50            project_id (str): ID of the Project
 51            share_id (str): ID of the Share
 52            max_items (int): Maximum number of records to get (default 10,000)
 53
 54        Example:
 55        ```python
 56        from cirro_api_client.v1.models import ShareType
 57        from cirro.cirro_client import CirroApi
 58
 59        cirro = CirroApi()
 60
 61        # List shares that are subscribed to
 62        subscribed_shares = cirro.shares.list(project_id="project-id", share_type=ShareType.SUBSCRIBER)
 63        cirro.datasets.list_shared("project-id", subscribed_shares[0].id)
 64        ```
 65        """
 66        return get_all_records(
 67            records_getter=lambda page_args: get_shared_datasets.sync(
 68                project_id=project_id,
 69                share_id=share_id,
 70                client=self._api_client,
 71                next_token=page_args.next_token,
 72                limit=page_args.limit
 73            ),
 74            max_items=max_items
 75        )
 76
 77    def import_public(self, project_id: str, import_request: ImportDataRequest) -> CreateResponse:
 78        """
 79        Download data from public repositories
 80
 81        Args:
 82            project_id (str): ID of the Project
 83            import_request (cirro_api_client.v1.models.ImportDataRequest):
 84
 85        Returns:
 86            ID of the created dataset
 87
 88        ```python
 89        from cirro_api_client.v1.models import ImportDataRequest, Tag
 90        from cirro.cirro_client import CirroApi
 91
 92        cirro = CirroApi()
 93        request = ImportDataRequest(
 94            name="Imported dataset",
 95            description="Description of the dataset",
 96            public_ids=["SRR123456", "SRR123457"],
 97            tags=[Tag(value="tag1")]
 98        )
 99        cirro.datasets.import_public("project-id", request)
100        ```
101        """
102        return import_public_dataset.sync(project_id=project_id, client=self._api_client, body=import_request)
103
104    def create(self, project_id: str, upload_request: UploadDatasetRequest) -> UploadDatasetCreateResponse:
105        """
106        Registers a dataset in Cirro, which can subsequently have files uploaded to it
107
108        Args:
109            project_id (str): ID of the Project
110            upload_request (cirro_api_client.v1.models.UploadDatasetRequest):
111
112        Returns:
113            ID of the created dataset and the path to upload files
114
115        ```python
116        from cirro_api_client.v1.models import UploadDatasetRequest, Tag
117        from cirro.cirro_client import CirroApi
118
119        cirro = CirroApi()
120        request = UploadDatasetRequest(
121            name="Name of new dataset",
122            process_id="paired_dnaseq",
123            expected_files=["read_1.fastq.gz", "read_2.fastq.gz"],
124            description="Description of the dataset",
125            tags=[Tag(value="tag1"), Tag(value="tag2")]
126        )
127        cirro.datasets.create("project-id", request)
128        ```
129        """
130        return upload_dataset.sync(project_id=project_id, client=self._api_client, body=upload_request)
131
132    def get(self, project_id: str, dataset_id: str) -> Optional[DatasetDetail]:
133        """
134        Gets detailed information about a dataset
135
136        Args:
137            project_id (str): ID of the Project
138            dataset_id (str): ID of the Dataset
139
140        Returns:
141            The dataset, if found
142        """
143        return get_dataset.sync(project_id=project_id, dataset_id=dataset_id, client=self._api_client)
144
145    def update(self, project_id: str, dataset_id: str, request: UpdateDatasetRequest) -> DatasetDetail:
146        """
147        Update info on a dataset (name, description, and/or tags)
148
149        Args:
150            project_id (str): ID of the Project
151            dataset_id (str): ID of the Dataset
152            request (cirro_api_client.v1.models.UpdateDatasetRequest):
153
154        Returns:
155            The updated dataset
156
157        ```python
158        from cirro_api_client.v1.models import UpdateDatasetRequest
159        from cirro.cirro_client import CirroApi
160
161        cirro = CirroApi()
162        request = UpdateDatasetRequest(
163            name="Name of new dataset",
164            process_id="paired_dnaseq",
165            description="Description of the dataset"
166        )
167        cirro.datasets.update("project-id", "dataset-id", request)
168        ```
169        """
170        return update_dataset.sync(project_id=project_id, dataset_id=dataset_id, body=request, client=self._api_client)
171
172    def delete(self, project_id: str, dataset_id: str) -> None:
173        """
174        Delete a dataset
175
176        After a dataset has been deleted, the files associated with that
177        dataset are saved according to the project's retention time.
178
179        Args:
180            project_id (str): ID of the Project
181            dataset_id (str): ID of the Dataset
182        """
183        delete_dataset.sync_detailed(project_id=project_id, dataset_id=dataset_id, client=self._api_client)
184
185    def get_assets_listing(self, project_id: str, dataset_id: str, file_limit: int = 100000) -> DatasetAssets:
186        """
187        Gets a listing of files, charts, and other assets available for the dataset
188
189        Args:
190            project_id (str): ID of the Project
191            dataset_id (str): ID of the Dataset
192            file_limit (int): Maximum number of files to get (default 100,000)
193        """
194        dataset = self.get(project_id, dataset_id)
195        if file_limit < 1:
196            raise ValueError("file_limit must be greater than 0")
197        all_files = []
198        file_offset = 0
199        domain = None
200        artifacts = None
201
202        while len(all_files) < file_limit:
203            manifest = get_dataset_manifest.sync(
204                project_id=project_id,
205                dataset_id=dataset_id,
206                file_offset=file_offset,
207                client=self._api_client
208            )
209            all_files.extend(manifest.files)
210            file_offset += len(manifest.files)
211
212            if not artifacts:
213                artifacts = manifest.artifacts
214
215            domain = manifest.domain
216            if len(all_files) >= manifest.total_files or len(manifest.files) == 0:
217                break
218
219        files = [
220            File.from_file_entry(
221                f,
222                project_id=project_id,
223                dataset=dataset,
224                domain=domain
225            )
226            for f in all_files
227        ]
228        artifacts = [
229            Artifact(
230                artifact_type=a.type_,
231                file=File.from_file_entry(
232                    FileEntry(a.path),
233                    project_id=project_id,
234                    dataset=dataset,
235                    domain=domain
236                )
237            )
238            for a in artifacts
239        ]
240        return DatasetAssets(files=files, artifacts=artifacts)
241
242    def upload_files(self,
243                     project_id: str,
244                     dataset_id: str,
245                     directory: PathLike,
246                     files: List[PathLike] = None,
247                     file_path_map: Dict[PathLike, str] = None) -> None:
248        """
249        Uploads files to a given dataset from the specified directory.
250
251        All files must be relative to the specified directory.
252        If files need to be flattened, or you are sourcing files from multiple directories,
253        please include `file_path_map` or call this method multiple times.
254
255        Args:
256            project_id (str): ID of the Project
257            dataset_id (str): ID of the Dataset
258            directory (str|Path): Path to directory
259            files (typing.List[str|Path]): List of paths to files within the directory,
260                must be the same type as directory.
261            file_path_map (typing.Dict[str|Path, str|Path]): Optional mapping of file paths to upload
262             from source path to destination path, used to "re-write" paths within the dataset.
263        ```python
264        from cirro.cirro_client import CirroApi
265        from cirro.file_utils import generate_flattened_file_map
266
267        cirro = CirroApi()
268
269        directory = "~/Downloads"
270        # Re-write file paths
271        file_map = {
272            "data1/file1.fastq.gz": "file1.fastq.gz",
273            "data2/file2.fastq.gz": "file2.fastq.gz",
274            "file3.fastq.gz": "new_file3.txt"
275        }
276
277        # Or you could automate the flattening
278        files = ["data1/file1.fastq.gz"]
279        file_map = generate_flattened_file_map(files)
280
281        cirro.datasets.upload_files(
282            project_id="project-id",
283            dataset_id="dataset-id",
284            directory=directory,
285            files=list(file_map.keys()),
286            file_path_map=file_map
287        )
288        ```
289        """
290        if file_path_map is None:
291            file_path_map = {}
292
293        dataset = self.get(project_id, dataset_id)
294
295        access_context = FileAccessContext.upload_dataset(
296            project_id=project_id,
297            dataset_id=dataset_id,
298            base_url=dataset.s3
299        )
300
301        self._file_service.upload_files(
302            access_context=access_context,
303            directory=directory,
304            files=files,
305            file_path_map=file_path_map
306        )
307
308    def validate_folder(
309        self,
310        project_id: str,
311        dataset_id: str,
312        local_folder: PathLike,
313        file_limit: int = 100000
314    ) -> DatasetValidationResponse:
315        """
316        Validates that the contents of a dataset match that of a local folder.
317        """
318        ds_files = self.get_assets_listing(project_id, dataset_id, file_limit=file_limit).files
319
320        local_folder = Path(local_folder)
321        if not local_folder.is_dir():
322            raise ValueError(f"{local_folder} is not a valid local folder")
323
324        # Keep track of files from the dataset which match by checksum, don't match, or are missing
325        ds_files_matching = []
326        ds_files_not_matching = []
327        ds_files_missing = []
328        ds_validate_failed = []
329        for ds_file in ds_files:
330            ds_file_path = ds_file.normalized_path
331            # Get the corresponding local file
332            local_file = local_folder / ds_file_path
333            if not local_file.exists():
334                ds_files_missing.append(ds_file_path)
335            else:
336                try:
337                    if self._file_service.is_valid_file(ds_file, local_file):
338                        ds_files_matching.append(ds_file_path)
339                    else:
340                        ds_files_not_matching.append(ds_file_path)
341                except RuntimeWarning as e:
342                    logger.warning(f"File validation failed: {e}")
343                    ds_validate_failed.append(ds_file_path)
344
345        # Find local files that are not in the dataset
346        local_file_paths = [
347            file.relative_to(local_folder).as_posix()
348            for file in local_folder.rglob("*")
349            if not file.is_dir() and not is_hidden_file(file)
350        ]
351        dataset_file_paths = [file.normalized_path for file in ds_files]
352        local_only_files = [
353            file
354            for file in local_file_paths
355            if file not in dataset_file_paths
356        ]
357
358        return DatasetValidationResponse(
359            files_matching=ds_files_matching,
360            files_not_matching=ds_files_not_matching,
361            files_missing=ds_files_missing,
362            local_only_files=local_only_files,
363            validate_errors=ds_validate_failed,
364        )
365
366    def download_files(
367        self,
368        project_id: str,
369        dataset_id: str,
370        download_location: str,
371        files: Union[List[File], List[str]] = None,
372        file_limit: int = 100000
373    ) -> None:
374        """
375        Downloads files from a dataset
376
377        The `files` argument is used to optionally specify a subset of files
378        to be downloaded. By default, all files are downloaded.
379
380        Args:
381            project_id (str): ID of the Project
382            dataset_id (str): ID of the Dataset
383            download_location (str): Local destination for downloaded files
384            files (typing.List[str]): Optional list of files to download
385            file_limit (int): Maximum number of files to get (default 100,000)
386        """
387        if files is None:
388            files = self.get_assets_listing(project_id, dataset_id, file_limit=file_limit).files
389
390        if len(files) == 0:
391            return
392
393        first_file = files[0]
394        if isinstance(first_file, File):
395            files = [file.relative_path for file in files]
396            access_context = first_file.access_context
397        else:
398            dataset = self.get(project_id, dataset_id)
399            if dataset.share:
400                access_context = FileAccessContext.download_shared_dataset(project_id=project_id,
401                                                                           dataset_id=dataset_id,
402                                                                           base_url=dataset.s3)
403            else:
404                access_context = FileAccessContext.download(project_id=project_id,
405                                                            base_url=dataset.s3)
406
407        self._file_service.download_files(access_context, download_location, files)
408
409    def update_samplesheet(
410        self,
411        project_id: str,
412        dataset_id: str,
413        samplesheet: str
414    ):
415        """
416        Updates a samplesheet on a dataset
417
418        Args:
419            project_id (str): ID of the Project
420            dataset_id (str): ID of the Dataset
421            samplesheet (str): Samplesheet contents to update (should be a CSV string)
422        """
423        dataset = self.get(project_id, dataset_id)
424        access_context = FileAccessContext.upload_sample_sheet(project_id=project_id,
425                                                               dataset_id=dataset_id,
426                                                               base_url=dataset.s3)
427
428        samplesheet_key = f'{access_context.prefix}/samplesheet.csv'
429        self._file_service.create_file(
430            access_context=access_context,
431            key=samplesheet_key,
432            contents=samplesheet,
433            content_type='text/csv'
434        )

 13class ExecutionService(BaseService):
 14    """
 15    Service for interacting with the Execution endpoints
 16    """
 17    def run_analysis(self, project_id: str, request: RunAnalysisRequest) -> CreateResponse:
 18        """
 19        Launch an analysis job running a process on a set of inputs
 20
 21        Args:
 22            project_id (str): ID of the Project
 23            request (cirro_api_client.v1.models.RunAnalysisRequest):
 24
 25        Returns:
 26            The ID of the created dataset
 27
 28        ```python
 29        from cirro_api_client.v1.models import RunAnalysisRequest, RunAnalysisRequestParams
 30        from cirro.cirro_client import CirroApi
 31
 32        # Example:
 33        # Run the "process-nf-core-rnaseq-3_8" process using input data
 34        # from a dataset with the id "source-dataset-id"
 35
 36        # Optional analysis parameters
 37        params = RunAnalysisRequestParams.from_dict({
 38            "param_a": "val_a",
 39            "param_b": "val_b"
 40        })
 41
 42        cirro = CirroApi()
 43        request = RunAnalysisRequest(
 44            name="Name of the newly created dataset",
 45            description="Longer description of newly created dataset",
 46            process_id="process-nf-core-rnaseq-3_8",
 47            source_dataset_ids=["source-dataset-id"],
 48            params=params
 49        )
 50        cirro.execution.run_analysis("project-id", request)
 51        ```
 52        """
 53
 54        form_spec = get_process_parameters.sync(
 55            process_id=request.process_id,
 56            client=self._api_client
 57        )
 58
 59        ParameterSpecification(
 60            form_spec
 61        ).validate_params(
 62            request.params.to_dict() if request.params else {}
 63        )
 64
 65        return run_analysis.sync(
 66            project_id=project_id,
 67            body=request,
 68            client=self._api_client
 69        )
 70
 71    def stop_analysis(self, project_id: str, dataset_id: str):
 72        """
 73        Terminates all jobs related to a running analysis
 74
 75        Args:
 76            project_id (str): ID of the Project
 77            dataset_id (str): ID of the Dataset
 78        """
 79
 80        return stop_analysis.sync(
 81            project_id=project_id,
 82            dataset_id=dataset_id,
 83            client=self._api_client
 84        )
 85
 86    def get_project_summary(self, project_id: str) -> Dict[str, List[Task]]:
 87        """
 88        Gets an overview of the executions currently running in the project, by job queue
 89
 90        Args:
 91            project_id (str): ID of the Project
 92
 93        Returns:
 94            `cirro_api_client.v1.models.GetProjectSummaryResponse200`
 95        """
 96
 97        return get_project_summary.sync(
 98            project_id=project_id,
 99            client=self._api_client
100        ).additional_properties
101
102    def get_execution_logs(self, project_id: str, dataset_id: str, force_live=False) -> str:
103        """
104        Gets live logs from main execution task
105
106        Args:
107            project_id (str): ID of the Project
108            dataset_id (str): ID of the Dataset
109            force_live (bool): If True, it will fetch logs from CloudWatch,
110                even if the execution is already completed
111
112        """
113
114        resp = get_execution_logs.sync(
115            project_id=project_id,
116            dataset_id=dataset_id,
117            force_live=force_live,
118            client=self._api_client
119        )
120
121        return '\n'.join(e.message for e in resp.events)
122
123    def get_tasks_for_execution(self, project_id: str, dataset_id: str, force_live=False) -> Optional[List[Task]]:
124        """
125        Gets the tasks submitted by the workflow execution
126
127        Args:
128            project_id (str): ID of the Project
129            dataset_id (str): ID of the Dataset
130            force_live (bool): If True, it will try to get the list of jobs
131                from the executor (i.e., AWS Batch), rather than the workflow report
132        """
133
134        return get_tasks_for_execution.sync(
135            project_id=project_id,
136            dataset_id=dataset_id,
137            force_live=force_live,
138            client=self._api_client
139        )
140
141    def get_task_logs(self, project_id: str, dataset_id: str, task_id: str, force_live=False) -> str:
142        """
143        Gets the log output from an individual task
144
145        Args:
146            project_id (str): ID of the Project
147            dataset_id (str): ID of the Dataset
148            task_id (str): ID of the task
149            force_live (bool): If True, it will fetch logs from CloudWatch,
150                even if the execution is already completed
151
152        """
153
154        resp = get_task_logs.sync(
155            project_id=project_id,
156            dataset_id=dataset_id,
157            task_id=task_id,
158            force_live=force_live,
159            client=self._api_client
160        )
161
162        return '\n'.join(e.message for e in resp.events)