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        )

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

 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)

10class ComputeEnvironmentService(BaseService):
11    """
12    Service for interacting with the Compute Environment endpoints
13    """
14
15    def list_environments_for_project(self, project_id: str) -> List[ComputeEnvironmentConfiguration]:
16        """
17        List of custom compute environments for a project (i.e., an agent)
18
19        Args:
20            project_id (str): Project ID
21
22        Returns:
23            List of compute environments that are available for the project
24        """
25        return get_compute_environments.sync(project_id=project_id,
26                                             client=self._api_client)

 18class FileService(BaseService):
 19    """
 20    Service for interacting with files
 21    """
 22    enable_additional_checksum: bool
 23    transfer_retries: int
 24    _get_token_lock = threading.Lock()
 25    _read_token_cache: Dict[str, AWSCredentials] = {}
 26
 27    def __init__(self, api_client, enable_additional_checksum, transfer_retries):
 28        """
 29        Instantiates the file service class
 30        """
 31        self._api_client = api_client
 32        self.enable_additional_checksum = enable_additional_checksum
 33        self.transfer_retries = transfer_retries
 34
 35    def get_access_credentials(self, access_context: FileAccessContext) -> AWSCredentials:
 36        """
 37        Retrieves credentials to access files
 38
 39        Args:
 40            access_context (cirro.models.file.FileAccessContext): File access context, use class methods to generate
 41        """
 42        access_request = access_context.file_access_request
 43
 44        if access_request.access_type == ProjectAccessType.PROJECT_DOWNLOAD:
 45            return self._get_project_read_credentials(access_context)
 46
 47        else:
 48            return generate_project_file_access_token.sync(
 49                client=self._api_client,
 50                project_id=access_context.project_id,
 51                body=access_context.file_access_request
 52            )
 53
 54    def _get_project_read_credentials(self, access_context: FileAccessContext):
 55        """
 56        Retrieves credentials to read project data, this can be cached
 57        """
 58        access_request = access_context.file_access_request
 59        project_id = access_context.project_id
 60        with self._get_token_lock:
 61            cached_token = self._read_token_cache.get(project_id)
 62
 63            if not cached_token or datetime.now(tz=timezone.utc) > cached_token.expiration:
 64                new_token = generate_project_file_access_token.sync(
 65                    client=self._api_client,
 66                    project_id=project_id,
 67                    body=access_request
 68                )
 69
 70                self._read_token_cache[project_id] = new_token
 71
 72        return self._read_token_cache[project_id]
 73
 74    def get_aws_s3_client(self, access_context: FileAccessContext) -> BaseClient:
 75        """
 76        Gets the underlying AWS S3 client to perform operations on files
 77
 78        This is seeded with refreshable credentials from the access_context parameter
 79
 80        This may be used to perform advanced operations, such as CopyObject, S3 Select, etc.
 81        """
 82        s3_client = self._generate_s3_client(access_context)
 83        return s3_client.get_aws_client()
 84
 85    def get_file(self, file: File) -> bytes:
 86        """
 87        Gets the contents of a file
 88
 89        Args:
 90            file (cirro.models.file.File):
 91
 92        Returns:
 93            The raw bytes of the file
 94        """
 95        return self.get_file_from_path(file.access_context, file.relative_path)
 96
 97    def get_file_from_path(self, access_context: FileAccessContext, file_path: str) -> bytes:
 98        """
 99        Gets the contents of a file by providing the path, used internally
100
101        Args:
102            access_context (cirro.models.file.FileAccessContext): File access context, use class methods to generate
103            file_path (str): Relative path to file within dataset
104
105        Returns:
106            The raw bytes of the file
107        """
108        s3_client = self._generate_s3_client(access_context)
109
110        full_path = f'{access_context.prefix}/{file_path}'.lstrip('/')
111
112        return s3_client.get_file(access_context.bucket, full_path)
113
114    def create_file(self, access_context: FileAccessContext, key: str,
115                    contents: str, content_type: str) -> None:
116        """
117        Creates a file at the specified path
118
119        Args:
120            access_context (cirro.models.file.FileAccessContext): File access context, use class methods to generate
121            key (str): Key of object to create
122            contents (str): Content of object
123            content_type (str):
124        """
125        s3_client = self._generate_s3_client(access_context)
126
127        s3_client.create_object(
128            key=key,
129            contents=contents,
130            content_type=content_type,
131            bucket=access_context.bucket
132        )
133
134    def upload_files(self,
135                     access_context: FileAccessContext,
136                     directory: PathLike,
137                     files: List[PathLike],
138                     file_path_map: Dict[PathLike, str]) -> None:
139        """
140        Uploads a list of files from the specified directory
141
142        Args:
143            access_context (cirro.models.file.FileAccessContext): File access context, use class methods to generate
144            directory (str|Path): Path to directory
145            files (typing.List[str|Path]): List of paths to files within the directory
146                must be the same type as directory.
147            file_path_map (typing.Dict[str|Path, str]): Optional mapping of file paths to upload
148             from source path to destination path, used to "re-write" paths within the dataset.
149        """
150        s3_client = self._generate_s3_client(access_context)
151
152        upload_directory(
153            directory=directory,
154            files=files,
155            file_path_map=file_path_map,
156            s3_client=s3_client,
157            bucket=access_context.bucket,
158            prefix=access_context.prefix,
159            max_retries=self.transfer_retries
160        )
161
162    def download_files(self, access_context: FileAccessContext, directory: str, files: List[str]) -> None:
163        """
164        Download a list of files to the specified directory
165
166        Args:
167            access_context (cirro.models.file.FileAccessContext): File access context, use class methods to generate
168            directory (str): download location
169            files (List[str]): relative path of files to download
170        """
171        s3_client = self._generate_s3_client(access_context)
172
173        download_directory(
174            directory,
175            files,
176            s3_client,
177            access_context.bucket,
178            access_context.prefix
179        )
180
181    def _generate_s3_client(self, access_context: FileAccessContext):
182        """
183        Generates the Cirro-S3 client to perform operations on files
184        """
185        return S3Client(
186            partial(self.get_access_credentials, access_context),
187            self.enable_additional_checksum
188        )