ops_utils.terra_util

  1"""Utilities for working with Terra."""
  2import json
  3import logging
  4import re
  5from typing import Any, Optional
  6import requests
  7import time
  8import zipfile
  9import os
 10
 11from . import deprecated
 12from .vars import GCP
 13from .gcp_utils import GCPCloudFunctions
 14from .request_util import GET, POST, PATCH, PUT, DELETE, RunRequest
 15
 16TERRA_LINK = "https://api.firecloud.org/api"
 17"""@private"""
 18LEONARDO_LINK = "https://leonardo.dsde-prod.broadinstitute.org/api"
 19"""@private"""
 20WORKSPACE_LINK = "https://workspace.dsde-prod.broadinstitute.org/api/workspaces/v1"
 21"""@private"""
 22SAM_LINK = "https://sam.dsde-prod.broadinstitute.org/api"
 23"""@private"""
 24RAWLS_LINK = "https://rawls.dsde-prod.broadinstitute.org/api"
 25"""@private"""
 26
 27MEMBER = "member"
 28ADMIN = "admin"
 29
 30
 31class Terra:
 32    """Class for generic Terra utilities."""
 33
 34    def __init__(self, request_util: RunRequest):
 35        """
 36        Initialize the Terra class.
 37
 38        **Args:**
 39        - request_util (`ops_utils.request_util.RunRequest`): An instance of a
 40            request utility class to handle HTTP requests.
 41        """
 42        self.request_util = request_util
 43        """@private"""
 44
 45    def fetch_accessible_workspaces(self, fields: Optional[list[str]]) -> requests.Response:
 46        """
 47        Fetch the list of accessible workspaces.
 48
 49        **Args:**
 50        - fields (list[str], optional): A list of fields to include in the response. If None, all fields are included.
 51
 52        **Returns:**
 53        - requests.Response: The response from the request.
 54        """
 55        fields_str = "fields=" + ",".join(fields) if fields else ""
 56        url = f'{RAWLS_LINK}/workspaces?{fields_str}'
 57        return self.request_util.run_request(
 58            uri=url,
 59            method=GET
 60        )
 61
 62    def get_pet_account_json(self) -> requests.Response:
 63        """
 64        Get the service account JSON.
 65
 66        **Returns:**
 67        - requests.Response: The response from the request.
 68        """
 69        url = f"{SAM_LINK}/google/v1/user/petServiceAccount/key"
 70        return self.request_util.run_request(
 71            uri=url,
 72            method=GET
 73        )
 74
 75
 76class TerraGroups:
 77    """A class to manage Terra groups and their memberships."""
 78
 79    GROUP_MEMBERSHIP_OPTIONS = [MEMBER, ADMIN]
 80    """@private"""
 81    CONFLICT_STATUS_CODE = 409
 82    """@private"""
 83
 84    def __init__(self, request_util: RunRequest):
 85        """
 86        Initialize the TerraGroups class.
 87
 88        **Args:**
 89        - request_util (`ops_utils.request_util.RunRequest`): An instance of a request
 90         utility class to handle HTTP requests.
 91        """
 92        self.request_util = request_util
 93        """@private"""
 94
 95    def _check_role(self, role: str) -> None:
 96        """
 97        Check if the role is valid.
 98
 99        Args:
100            role (str): The role to check.
101
102        Raises:
103            ValueError: If the role is not one of the allowed options.
104        """
105        if role not in self.GROUP_MEMBERSHIP_OPTIONS:
106            raise ValueError(f"Role must be one of {self.GROUP_MEMBERSHIP_OPTIONS}")
107
108    def remove_user_from_group(self, group: str, email: str, role: str) -> requests.Response:
109        """
110        Remove a user from a group.
111
112        **Args:**
113        - group (str): The name of the group.
114        - email (str): The email of the user to remove.
115        - role (str): The role of the user in the group
116            (must be one of `ops_utils.terra_utils.MEMBER` or `ops_utils.terra_utils.ADMIN`).
117
118        **Returns:**
119        - requests.Response: The response from the request.
120        """
121        url = f"{SAM_LINK}/groups/v1/{group}/{role}/{email}"
122        self._check_role(role)
123        logging.info(f"Removing {email} from group {group}")
124        return self.request_util.run_request(
125            uri=url,
126            method=DELETE
127        )
128
129    def create_group(self, group_name: str, continue_if_exists: bool = False) -> requests.Response:
130        """
131        Create a new group.
132
133        **Args:**
134        - group_name (str): The name of the group to create.
135        - continue_if_exists (bool, optional): Whether to continue if the group already exists. Defaults to `False`.
136
137        **Returns:**
138        - requests.Response: The response from the request.
139        """
140        url = f"{SAM_LINK}/groups/v1/{group_name}"
141        accept_return_codes = [self.CONFLICT_STATUS_CODE] if continue_if_exists else []
142        response = self.request_util.run_request(
143            uri=url,
144            method=POST,
145            accept_return_codes=accept_return_codes
146        )
147        if continue_if_exists and response.status_code == self.CONFLICT_STATUS_CODE:
148            logging.info(f"Group {group_name} already exists. Continuing.")
149            return response
150        else:
151            logging.info(f"Created group {group_name}")
152            return response
153
154    def delete_group(self, group_name: str) -> requests.Response:
155        """
156        Delete a group.
157
158        **Args:**
159        - group_name (str): The name of the group to delete.
160
161        **Returns:**
162        - requests.Response: The response from the request.
163        """
164        url = f"{SAM_LINK}/groups/v1/{group_name}"
165        logging.info(f"Deleting group {group_name}")
166        return self.request_util.run_request(
167            uri=url,
168            method=DELETE
169        )
170
171    def add_user_to_group(
172            self, group: str, email: str, role: str, continue_if_exists: bool = False
173    ) -> requests.Response:
174        """
175        Add a user to a group.
176
177        **Args:**
178        - group (str): The name of the group.
179        - email (str): The email of the user to add.
180        - role (str): The role of the user in the group
181            (must be one of `ops_utils.terra_utils.MEMBER` or `ops_utils.terra_utils.ADMIN`).
182        - continue_if_exists (bool, optional): Whether to continue if the user is already in the group.
183                Defaults to `False`.
184
185        **Returns:**
186        - requests.Response: The response from the request.
187        """
188        url = f"{SAM_LINK}/groups/v1/{group}/{role}/{email}"
189        self._check_role(role)
190        accept_return_codes = [self.CONFLICT_STATUS_CODE] if continue_if_exists else []
191        logging.info(f"Adding {email} to group {group} as {role}")
192        return self.request_util.run_request(
193            uri=url,
194            method=PUT,
195            accept_return_codes=accept_return_codes
196        )
197
198
199class TerraWorkspace:
200    """Terra workspace class to manage workspaces and their attributes."""
201
202    CONFLICT_STATUS_CODE = 409
203    """@private"""
204
205    def __init__(self, billing_project: str, workspace_name: str, request_util: RunRequest):
206        """
207        Initialize the TerraWorkspace class.
208
209        **Args:**
210        - billing_project (str): The billing project associated with the workspace.
211        - workspace_name (str): The name of the workspace.
212        - request_util (`ops_utils.request_util.RunRequest`): An instance of a
213            request utility class to handle HTTP requests.
214        """
215        self.billing_project = billing_project
216        """@private"""
217        self.workspace_name = workspace_name
218        """@private"""
219        self.workspace_id = None
220        """@private"""
221        self.resource_id = None
222        """@private"""
223        self.storage_container = None
224        """@private"""
225        self.bucket = None
226        """@private"""
227        self.wds_url = None
228        """@private"""
229        self.account_url: Optional[str] = None
230        """@private"""
231        self.request_util = request_util
232        """@private"""
233
234    def __repr__(self) -> str:
235        """
236        Return a string representation of the TerraWorkspace instance.
237
238        Returns:
239            str: The string representation of the TerraWorkspace instance.
240        """
241        return f"{self.billing_project}/{self.workspace_name}"
242
243    def _yield_all_entity_metrics(self, entity: str, total_entities_per_page: int = 40000) -> Any:
244        """
245        Yield all entity metrics from the workspace.
246
247        Args:
248            entity (str): The type of entity to query.
249            total_entities_per_page (int, optional): The number of entities per page. Defaults to 40000.
250
251        Yields:
252            Any: The JSON response containing entity metrics.
253        """
254        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/entityQuery/{entity}?pageSize={total_entities_per_page}"  # noqa: E501
255        response = self.request_util.run_request(
256            uri=url,
257            method=GET,
258            content_type="application/json"
259        )
260        first_page_json = response.json()
261        yield first_page_json
262        total_pages = first_page_json["resultMetadata"]["filteredPageCount"]
263        logging.info(
264            f"Looping through {total_pages} pages of data")
265
266        for page in range(2, total_pages + 1):
267            logging.info(f"Getting page {page} of {total_pages}")
268            next_page = self.request_util.run_request(
269                uri=url,
270                method=GET,
271                content_type="application/json",
272                params={"page": page}
273            )
274            yield next_page.json()
275
276    @staticmethod
277    def validate_terra_headers_for_tdr_conversion(table_name: str, headers: list[str]) -> None:
278        """Check that all headers follow the standards required by TDR.
279
280        **Args:**
281        - table_name (str): The name of the Terra table.
282        - headers (list[str]): The headers of the Terra table to validate.
283
284        **Raises:**
285        - ValueError if any headers are considered invalid by TDR standards
286        """
287        tdr_header_allowed_pattern = "^[a-zA-Z][_a-zA-Z0-9]*$"
288        tdr_max_header_length = 63
289
290        headers_containing_too_many_characters = []
291        headers_contain_invalid_characters = []
292
293        for header in headers:
294            if len(header) > tdr_max_header_length:
295                headers_containing_too_many_characters.append(header)
296            if not re.match(tdr_header_allowed_pattern, header):
297                headers_contain_invalid_characters.append(header)
298
299        base_error_message = """In order to proceed, please update the problematic header(s) in you Terra table,
300        and then re-attempt the import once all problematic header(s) have been updated to follow TDR rules for
301        header naming."""
302        too_many_characters_error_message = f"""The following header(s) in table "{table_name}" contain too many
303        characters: "{', '.join(headers_containing_too_many_characters)}". The max number of characters for a header
304        allowed in TDR is {tdr_max_header_length}.\n"""
305        invalid_characters_error_message = f"""The following header(s) in table "{table_name}" contain invalid
306        characters: "{', '.join(headers_contain_invalid_characters)}". TDR headers must start with a letter, and must
307        only contain numbers, letters, and underscore characters.\n"""
308
309        error_to_report = ""
310        if headers_containing_too_many_characters:
311            error_to_report += too_many_characters_error_message
312        if headers_contain_invalid_characters:
313            error_to_report += invalid_characters_error_message
314        if error_to_report:
315            error_to_report += base_error_message
316            raise ValueError(error_to_report)
317
318    def get_workspace_info(self) -> requests.Response:
319        """
320        Get workspace information.
321
322        **Returns:**
323        - requests.Response: The response from the request.
324        """
325        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}"
326        logging.info(
327            f"Getting workspace info for {self.billing_project}/{self.workspace_name}")
328        return self.request_util.run_request(uri=url, method=GET)
329
330    def get_gcp_workspace_metrics(self, entity_type: str, remove_dicts: bool = False) -> list[dict]:
331        """
332        Get metrics for a specific entity type in the workspace (specifically for Terra on GCP).
333
334        **Args:**
335        - entity_type (str): The type of entity to get metrics for.
336        - remove_dicts (bool, optional): Whether to remove dictionaries from the workspace metrics. Defaults to `False`.
337
338        **Returns:**
339        - list[dict]: A list of dictionaries containing entity metrics.
340        """
341        results = []
342        logging.info(f"Getting {entity_type} metadata for {self.billing_project}/{self.workspace_name}")
343
344        for page in self._yield_all_entity_metrics(entity=entity_type):
345            results.extend(page["results"])
346
347        # If remove_dicts is True, remove dictionaries from the workspace metrics
348        if remove_dicts:
349            for row in results:
350                row['attributes'] = self._remove_dict_from_attributes(row['attributes'])
351        return results
352
353    def get_specific_entity_metrics(self, entity_type: str, entity_name: str) -> requests.Response:
354        """
355        Get specific entity metrics for a given entity type and name.
356
357        **Args:**
358        - entity_type (str): The type of entity to get metrics for.
359        - entity_name (str): The name of the entity to get metrics for.
360
361        **Returns:**
362        - requests.Response: The response from the request.
363        """
364        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/entities/{entity_type}/{entity_name}"
365        return self.request_util.run_request(uri=url, method=GET)
366
367    def _remove_dict_from_attributes(self, attributes: dict) -> dict:
368        """
369        Remove dictionaries from the attributes.
370
371        Args:
372            attributes (dict): The attributes to remove dictionaries from.
373
374        Returns:
375            dict: The updated attributes with no dictionaries.
376        """
377        for key, value in attributes.items():
378            attributes[key] = self._remove_dict_from_cell(value)
379        return attributes
380
381    def _remove_dict_from_cell(self, cell_value: Any) -> Any:
382        """
383        Remove a dictionary from a cell.
384
385        Args:
386            cell_value (Any): The dictionary to remove.
387
388        Returns:
389            Any: The updated cell with no dictionaries.
390        """
391        if isinstance(cell_value, dict):
392            entity_name = cell_value.get("entityName")
393            # If the cell value is a dictionary, check if it has an entityName key
394            if entity_name:
395                # If the cell value is a dictionary with an entityName key, return the entityName
396                return entity_name
397            entity_list = cell_value.get("items")
398            if entity_list or entity_list == []:
399                # If the cell value is a list of dictionaries, recursively call this function on each dictionary
400                return [
401                    self._remove_dict_from_cell(entity) for entity in entity_list
402                ]
403            return cell_value
404        return cell_value
405
406    def get_workspace_bucket(self) -> str:
407        """
408        Get the workspace bucket name. Does not include the `gs://` prefix.
409
410        **Returns:**
411        - str: The bucket name.
412        """
413        return self.get_workspace_info().json()["workspace"]["bucketName"]
414
415    def get_workspace_entity_info(self, use_cache: bool = True) -> requests.Response:
416        """
417        Get workspace entity information.
418
419        **Args:**
420        - use_cache (bool, optional): Whether to use cache. Defaults to `True`.
421
422        **Returns:**
423        - requests.Response: The response from the request.
424        """
425        use_cache = "true" if use_cache else "false"  # type: ignore[assignment]
426        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/entities?useCache={use_cache}"
427        return self.request_util.run_request(uri=url, method=GET)
428
429    def get_workspace_acl(self) -> requests.Response:
430        """
431        Get the workspace access control list (ACL).
432
433        **Returns:**
434        - requests.Response: The response from the request.
435        """
436        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/acl"
437        return self.request_util.run_request(
438            uri=url,
439            method=GET
440        )
441
442    def update_user_acl(
443            self,
444            email: str,
445            access_level: str,
446            can_share: bool = False,
447            can_compute: bool = False,
448            invite_users_not_found: bool = False,
449    ) -> requests.Response:
450        """
451        Update the access control list (ACL) for a user in the workspace.
452
453        **Args:**
454        - email (str): The email of the user.
455        - access_level (str): The access level to grant to the user.
456        - can_share (bool, optional): Whether the user can share the workspace. Defaults to `False`.
457        - can_compute (bool, optional): Whether the user can compute in the workspace. Defaults to `False`.
458        - invite_users_not_found (bool, optional): Whether a user that's not found should still be invited to access
459                the workspace. Defaults to `False`
460
461        **Returns:**
462        - requests.Response: The response from the request.
463        """
464        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/acl?" + \
465              f"inviteUsersNotFound={str(invite_users_not_found).lower()}"
466        payload = {
467            "email": email,
468            "accessLevel": access_level,
469            "canShare": can_share,
470            "canCompute": can_compute,
471        }
472        logging.info(
473            f"Updating user {email} to {access_level} in workspace {self.billing_project}/{self.workspace_name}")
474        response = self.request_util.run_request(
475            uri=url,
476            method=PATCH,
477            content_type="application/json",
478            data="[" + json.dumps(payload) + "]"
479        )
480
481        if response.json()["usersNotFound"] and not invite_users_not_found:
482            # Will be a list of one user
483            user_not_found = response.json()["usersNotFound"][0]
484            raise Exception(
485                f'The user {user_not_found["email"]} was not found and access was not updated'
486            )
487        return response
488
489    @deprecated(
490        """Firecloud functionality has been sunset. There is currently no support for adding library attributes in Terra."""  # noqa: E501
491    )
492    def put_metadata_for_library_dataset(self, library_metadata: dict, validate: bool = False) -> requests.Response:
493        """
494        Update the metadata for a library dataset.
495
496        **Args:**
497        - library_metadata (dict): The metadata to update.
498        - validate (bool, optional): Whether to validate the metadata. Defaults to `False`.
499
500        **Returns:**
501        - requests.Response: The response from the request.
502        """
503        acl = f"{TERRA_LINK}/library/{self.billing_project}/{self.workspace_name}" + \
504              f"/metadata?validate={str(validate).lower()}"
505        return self.request_util.run_request(
506            uri=acl,
507            method=PUT,
508            data=json.dumps(library_metadata)
509        )
510
511    def update_multiple_users_acl(
512            self, acl_list: list[dict], invite_users_not_found: bool = False
513    ) -> requests.Response:
514        """
515        Update the access control list (ACL) for multiple users in the workspace.
516
517        **Args:**
518        - acl_list (list[dict]): A list of dictionaries containing the ACL information for each user.
519        - invite_users_not_found (bool, optional): Whether a user that's not found should still be invited to access
520                the workspace. Defaults to `False`
521
522        **Returns:**
523        - requests.Response: The response from the request.
524        """
525        url = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/acl?" + \
526            f"inviteUsersNotFound={str(invite_users_not_found).lower()}"
527        logging.info(
528            f"Updating users in workspace {self.billing_project}/{self.workspace_name}")
529        response = self.request_util.run_request(
530            uri=url,
531            method=PATCH,
532            content_type="application/json",
533            data=json.dumps(acl_list)
534        )
535
536        if response.json()["usersNotFound"] and not invite_users_not_found:
537            # Will be a list of one user
538            users_not_found = [u["email"] for u in response.json()["usersNotFound"]]
539            raise Exception(
540                f"The following users were not found and access was not updated: {users_not_found}"
541            )
542        return response
543
544    def create_workspace(
545            self,
546            auth_domain: list[dict] = [],
547            attributes: dict = {},
548            continue_if_exists: bool = False,
549    ) -> requests.Response:
550        """
551        Create a new workspace in Terra.
552
553        **Args:**
554        - auth_domain (list[dict], optional): A list of authorization domains. Should look
555                like `[{"membersGroupName": "some_auth_domain"}]`. Defaults to an empty list.
556        - attributes (dict, optional): A dictionary of attributes for the workspace. Defaults to an empty dictionary.
557        - continue_if_exists (bool, optional): Whether to continue if the workspace already exists. Defaults to `False`.
558
559        **Returns:**
560        - requests.Response: The response from the request.
561        """
562        payload = {
563            "namespace": self.billing_project,
564            "name": self.workspace_name,
565            "authorizationDomain": auth_domain,
566            "attributes": attributes,
567            "cloudPlatform": GCP
568        }
569        # If workspace already exists then continue if exists
570        accept_return_codes = [self.CONFLICT_STATUS_CODE] if continue_if_exists else []
571        logging.info(f"Creating workspace {self.billing_project}/{self.workspace_name}")
572        response = self.request_util.run_request(
573            uri=f"{TERRA_LINK}/workspaces",
574            method=POST,
575            content_type="application/json",
576            data=json.dumps(payload),
577            accept_return_codes=accept_return_codes
578        )
579        if continue_if_exists and response.status_code == self.CONFLICT_STATUS_CODE:
580            logging.info(f"Workspace {self.billing_project}/{self.workspace_name} already exists")
581        return response
582
583    def create_workspace_attributes_ingest_dict(self, workspace_attributes: Optional[dict] = None) -> list[dict]:
584        """
585        Create an ingest dictionary for workspace attributes.
586
587        **Args:**
588        - workspace_attributes (dict, optional): A dictionary of workspace attributes. Defaults to None.
589
590        **Returns:**
591        - list[dict]: A list of dictionaries containing the workspace attributes.
592        """
593        # If not provided then call API to get it
594        workspace_attributes = (
595            workspace_attributes if workspace_attributes
596            else self.get_workspace_info().json()["workspace"]["attributes"]
597        )
598
599        ingest_dict = []
600        for key, value in workspace_attributes.items():
601            # If value is dict just use 'items' as value
602            if isinstance(value, dict):
603                value = value.get("items")
604            # If value is list convert to comma separated string
605            if isinstance(value, list):
606                value = ", ".join(value)
607            ingest_dict.append(
608                {
609                    "attribute": key,
610                    "value": str(value) if value else None
611                }
612            )
613        return ingest_dict
614
615    def upload_metadata_to_workspace_table(self, entities_tsv: str) -> requests.Response:
616        """
617        Upload metadata to the workspace table.
618
619        **Args:**
620        - entities_tsv (str): The path to the TSV file containing the metadata to upload.
621
622        **Returns:**
623        - requests.Response: The response from the request.
624        """
625        endpoint = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/flexibleImportEntities"
626        data = {"entities": open(entities_tsv, "rb")}
627        return self.request_util.upload_file(
628            uri=endpoint,
629            data=data
630        )
631
632    def get_workspace_workflows(self) -> requests.Response:
633        """
634        Get the workflows for the workspace.
635
636        **Returns:**
637        - requests.Response: The response from the request.
638        """
639        uri = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/methodconfigs?allRepos=true"
640        return self.request_util.run_request(
641            uri=uri,
642            method=GET
643        )
644
645    def import_workflow(self, workflow_dict: dict, continue_if_exists: bool = False) -> requests.Response:
646        """
647        Import a workflow into the workspace.
648
649        **Args:**
650        - workflow_dict (dict): The dictionary containing the workflow information.
651        - continue_if_exists (bool, optional): Whether to continue if the workflow
652                already exists. Defaults to `False`.
653
654        **Returns:**
655        - requests.Response: The response from the request.
656        """
657        uri = f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/methodconfigs"
658        workflow_json = json.dumps(workflow_dict)
659        accept_return_codes = [self.CONFLICT_STATUS_CODE] if continue_if_exists else []
660        return self.request_util.run_request(
661            uri=uri,
662            method=POST,
663            data=workflow_json,
664            content_type="application/json",
665            accept_return_codes=accept_return_codes
666        )
667
668    def delete_workspace(self) -> requests.Response:
669        """
670        Delete a Terra workspace.
671
672        **Returns:**
673        - requests.Response: The response from the request.
674        """
675        return self.request_util.run_request(
676            uri=f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}",
677            method=DELETE
678        )
679
680    def update_workspace_attributes(self, attributes: list[dict]) -> requests.Response:
681        """
682        Update the attributes for the workspace.
683
684        **Args:**
685        - attributes (dict): The attributes to update.
686
687        **Returns:**
688        - requests.Response: The response from the request.
689        """
690        return self.request_util.run_request(
691            uri=f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/updateAttributes",
692            method=PATCH,
693            data=json.dumps(attributes),
694            content_type="application/json"
695        )
696
697    def leave_workspace(
698            self, workspace_id: Optional[str] = None, ignore_direct_access_error: bool = False
699    ) -> requests.Response:
700        """
701        Leave a workspace. If workspace ID not supplied, will look it up.
702
703        **Args:**
704        - workspace_id (str, optional): The workspace ID. Defaults to None.
705        - ignore_direct_access_error (Optional[bool], optional): Whether to ignore direct access errors.
706             Defaults to `False`.
707
708        **Returns:**
709        - requests.Response: The response from the request.
710        """
711        if not workspace_id:
712            workspace_info = self.get_workspace_info().json()
713            workspace_id = workspace_info['workspace']['workspaceId']
714        accepted_return_code = [403] if ignore_direct_access_error else []
715
716        res = self.request_util.run_request(
717            uri=f"{SAM_LINK}/resources/v2/workspace/{workspace_id}/leave",
718            method=DELETE,
719            accept_return_codes=accepted_return_code
720        )
721        if (res.status_code == 403
722                and res.json()["message"] == "You can only leave a resource that you have direct access to."):
723            logging.info(
724                f"Did not remove user from workspace with id '{workspace_id}' as current user does not have direct"
725                f"access to the workspace (they could be an owner on the billing project)"
726            )
727        return res
728
729    def change_workspace_public_setting(self, public: bool) -> requests.Response:
730        """
731        Change a workspace's public setting.
732
733        **Args:**
734        - public (bool, optional): Whether the workspace should be public. Set to `True` to be made
735         public, `False` otherwise.
736
737        **Returns:**
738        - requests.Response: The response from the request.
739        """
740        body = [
741            {
742                "settingType": "PubliclyReadable",
743                "config": {
744                    "enabled": public
745                }
746            }
747        ]
748        return self.request_util.run_request(
749            uri=f"{RAWLS_LINK}/workspaces/v2/{self.billing_project}/{self.workspace_name}/settings",
750            method=PUT,
751            content_type="application/json",
752            data=json.dumps(body)
753        )
754
755    def check_workspace_public(self, bucket: Optional[str] = None) -> requests.Response:
756        """
757        Check if a workspace is public.
758
759        **Args:**
760        - bucket (str, optional): The bucket name (provided without the `gs://` prefix). Will look
761        it up if not provided. Defaults to None.
762
763        **Returns:**
764        - requests.Response: The response from the request.
765        """
766        workspace_bucket = bucket if bucket else self.get_workspace_bucket()
767        bucket_prefix_stripped = workspace_bucket.removeprefix("fc-secure-").removeprefix("fc-")
768        return self.request_util.run_request(
769            uri=f"{SAM_LINK}/resources/v2/workspace/{bucket_prefix_stripped}/policies/reader/public",
770            method=GET
771        )
772
773    def delete_entity_table(self, entity_to_delete: str) -> requests.Response:
774        """Delete an entire entity table from a Terra workspace.
775
776        **Args:**
777        - entity_to_delete (str): The name of the entity table to delete.
778
779        **Returns:**
780        - requests.Response: The response from the request.
781        """
782        response = self.request_util.run_request(
783            uri=f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/entityTypes/{entity_to_delete}",
784            method=DELETE
785        )
786        if response.status_code == 204:
787            logging.info(
788                f"Successfully deleted entity table: '{entity_to_delete}' from workspace: "
789                f"'{self.billing_project}/{self.workspace_name}'"
790            )
791        else:
792            logging.error(
793                f"Encountered the following error while attempting to delete '{entity_to_delete}' "
794                f"table: {response.text}"
795            )
796        return response
797
798    def save_entity_table_version(self, entity_type: str, version_name: str) -> None:
799        """Save an entity table version in a Terra workspace.
800
801        **Args:**
802        - entity_type (str): The name of the entity table to save a new version for
803        - version_name (str): The name of the new version
804        """
805        # Get the workspace metrics
806        workspace_metrics = self.get_gcp_workspace_metrics(entity_type=entity_type)
807        file_name = f"{entity_type}.json"
808        # Write the workspace metrics to a JSON file
809        with open(file_name, "w") as json_file:
810            json.dump(workspace_metrics, json_file)
811
812        # Create a zip file with the same naming convention that Terra backend uses
813        timestamp_ms = int(time.time() * 1000)
814        zip_file_name = f"{entity_type}.v{timestamp_ms}.zip"
815        with zipfile.ZipFile(zip_file_name, "w", zipfile.ZIP_DEFLATED) as zipf:
816            zipf.write(file_name, arcname=f"json/{file_name}")
817
818        # Upload the zip file to subdirectory within the workspace's bucket (where Terra expects it to live)
819        workspace_info = self.get_workspace_info().json()
820        path_to_upload_to = os.path.join(
821            "gs://", workspace_info["workspace"]["bucketName"], ".data-table-versions", entity_type, zip_file_name
822        )
823        gcp_util = GCPCloudFunctions(project=workspace_info["workspace"]["googleProject"])
824        # Attempt to get the currently active gcloud account. Default to the workspace creator if that fails
825        try:
826            active_account = gcp_util.get_active_gcloud_account()
827        except Exception as e:
828            active_account = workspace_info["workspace"]["createdBy"]
829            logging.error(
830                f"Encountered the following exception while attempting to get current GCP account: {e}. "
831                f"Will set the owner of the new metadata version as the workspace creator instead."
832            )
833        gcp_util.upload_blob(
834            source_file=zip_file_name,
835            destination_path=path_to_upload_to,
836            custom_metadata={
837                "createdBy": active_account,
838                "entityType": entity_type,
839                "timestamp": timestamp_ms,
840                "description": version_name,
841            }
842        )
843
844    def add_user_comment_to_submission(self, submission_id: str, user_comment: str) -> requests.Response:
845        """
846        Add a user comment to a submission in Terra.
847
848        **Args:**
849        - submission_id (str): The ID of the submission to add a comment to.
850        - user_comment (str): The comment to add to the submission.
851
852        **Returns:**
853        - requests.Response: The response from the request.
854        """
855        logging.info(f"Attempting to add user comment: '{user_comment}' to submission: '{submission_id}'")
856        return self.request_util.run_request(
857            uri=f"{RAWLS_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/submissions/{submission_id}",
858            method=PATCH,
859            content_type="application/json",
860            data=json.dumps({"userComment": user_comment}),
861        )
862
863    def initiate_submission(
864            self,
865            method_config_namespace: str,
866            method_config_name: str,
867            entity_type: str,
868            entity_name: str,
869            expression: str,
870            user_comment: Optional[str],
871            use_call_cache: bool = True
872    ) -> requests.Response:
873        """
874        Initiate a submission within a Terra workspace.
875
876        Note - the workflow being initiated MUST already be imported into the workspace.
877
878        **Args:**
879        - method_config_namespace (str): The namespace of the method configuration.
880        - method_config_name (str): The name of the method configuration to use for the submission
881        (i.e. the workflow name).
882        - entity_type (str): The entity type to be used as input to the workflow (e.x. "sample", or "sample_set").
883        - entity_name (str): The name of the entity to be used as input to the workflow (e.x. "sample_1", or
884        "sample_set_1").
885        - expression (str): The "expression" to use. For example, if the `entity_type` is `sample` and the workflow is
886        launching one sample, this can be left as `this`. If the `entity_type` is `sample_set`, but one workflow should
887        be launched PER SAMPLE, the expression should be `this.samples`.
888        - user_comment (str, optional): The user comment to add to the submission.
889        - use_call_cache (bool, optional): Whether to use the call caching. Defaults to `True`.
890
891        **Returns:**
892        - requests.Response: The response from the request.
893        """
894        payload = {
895            "methodConfigurationNamespace": method_config_namespace,
896            "methodConfigurationName": method_config_name,
897            "entityType": entity_type,
898            "entityName": entity_name,
899            "expression": expression,
900            "useCallCache": use_call_cache,
901            "deleteIntermediateOutputFiles": False,
902            "useReferenceDisks": False,
903            "ignoreEmptyOutputs": False,
904        }
905        if user_comment:
906            payload["userComment"] = user_comment
907
908        return self.request_util.run_request(
909            uri=f"{TERRA_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/submissions",
910            method=POST,
911            content_type="application/json",
912            data=json.dumps(payload),
913        )
914      
915    def retry_failed_submission(self, submission_id: str) -> requests.Response:
916        """
917        Retry a failed submission in Terra.
918
919        **Args:**
920        - submission_id (str): The ID of the submission to retry.
921
922        **Returns:**
923        - requests.Response: The response from the request.
924        """
925        url = f"{RAWLS_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/submissions/{submission_id}/retry"
926        payload = {"retryType": "Failed"}
927        logging.info(
928            f"Retrying failed submission: '{submission_id}' in workspace {self.billing_project}/{self.workspace_name}"
929        )
930        return self.request_util.run_request(
931            uri=url,
932            method=POST,
933            content_type="application/json",
934            data=json.dumps(payload)
935        )
936
937    def get_submission_status(self, submission_id: str) -> requests.Response:
938        """
939        Get the status of a submission in Terra.
940
941        **Args:**
942        - submission_id (str): The ID of the submission.
943
944        **Returns:**
945        - requests.Response: The response from the request.
946        """
947        url = f"{RAWLS_LINK}/workspaces/{self.billing_project}/{self.workspace_name}/submissions/{submission_id}"
948        logging.info(
949            f"Getting status for submission: '{submission_id}' in workspace {self.billing_project}/{self.workspace_name}"
950        )
951        return self.request_util.run_request(
952            uri=url,
953            method=GET
954        )