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        )

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

 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)