ops_utils.tdr_utils.tdr_ingest_utils

  1"""Classes and functions for ingesting data into TDR."""
  2
  3import json
  4import logging
  5import sys
  6import pytz
  7from datetime import datetime
  8import math
  9from typing import Optional, Any
 10from dateutil import parser
 11
 12from ..vars import ARG_DEFAULTS
 13
 14from .tdr_api_utils import TDR, FilterOutSampleIdsAlreadyInDataset
 15from .tdr_job_utils import MonitorTDRJob
 16from ..terra_util import TerraWorkspace
 17
 18
 19class BatchIngest:
 20    """A class to handle batch ingestion of metadata into TDR (Terra Data Repository)."""
 21
 22    def __init__(
 23            self,
 24            ingest_metadata: list[dict],
 25            tdr: TDR,
 26            target_table_name: str,
 27            dataset_id: str,
 28            batch_size: int,
 29            bulk_mode: bool,
 30            update_strategy: str = "replace",
 31            waiting_time_to_poll: int = ARG_DEFAULTS["waiting_time_to_poll"],  # type: ignore[assignment]
 32            test_ingest: bool = False,
 33            load_tag: Optional[str] = None,
 34            file_to_uuid_dict: Optional[dict] = None,
 35            schema_info: Optional[dict] = None,
 36            skip_reformat: bool = False
 37    ):
 38        """
 39        Initialize the BatchIngest class.
 40
 41        **Args:**
 42        - ingest_metadata (list[dict]): The metadata to be ingested.
 43        - tdr (`ops_utils.tdr_utils.tdr_api_utils.TDR`): An instance of the TDR class.
 44        - target_table_name (str): The name of the target table.
 45        - dataset_id (str): The ID of the dataset.
 46        - batch_size (int): The size of each batch for ingestion.
 47        - bulk_mode (bool): Flag indicating if bulk mode should be used.
 48        - update_strategy (str, optional): The strategy for updating existing records. Defaults to `replace`.
 49        - waiting_time_to_poll (int, optional): The time to wait between polling for job status. Defaults to `90`.
 50        - test_ingest (bool, optional): Flag indicating if only the first batch should be
 51                ingested for testing. Defaults to `False`.
 52        - load_tag (str, optional): A tag to identify the load. Used so future ingests
 53                can pick up where left off. Defaults to None.
 54        - file_to_uuid_dict (dict, optional): Only useful for self-hosted dataset. Can get from
 55            create_file_uuid_dict_for_ingest_for_experimental_self_hosted_dataset. A dictionary mapping
 56            source file paths to UUIDs. If used will make ingest much quicker since no ingest
 57            or look up of file needed. Defaults to None.
 58        - schema_info (dict, optional): Schema information for the tables. Validates ingest data matches up
 59                with schema info. Defaults to None.
 60        - skip_reformat (bool, optional): Flag indicating if reformatting should be skipped. Defaults to `False`.
 61        """
 62        self.ingest_metadata = self._reformat_for_type_consistency(ingest_metadata)
 63        """@private"""
 64        self.tdr = tdr
 65        """@private"""
 66        self.target_table_name = target_table_name
 67        """@private"""
 68        self.dataset_id = dataset_id
 69        """@private"""
 70        self.batch_size = int(batch_size)
 71        """@private"""
 72        self.update_strategy = update_strategy
 73        """@private"""
 74        self.bulk_mode = bulk_mode
 75        """@private"""
 76        self.waiting_time_to_poll = waiting_time_to_poll
 77        """@private"""
 78        # Used if you want to run first batch and then exit after success
 79        self.test_ingest = test_ingest
 80        """@private"""
 81        self.load_tag = load_tag
 82        """@private"""
 83        self.file_to_uuid_dict = file_to_uuid_dict
 84        """@private"""
 85        # Used if you want to provide schema info for tables to make sure values match.
 86        # Should be dict with key being column name and value being dict with datatype
 87        self.schema_info = schema_info
 88        """@private"""
 89        # Use if input is already formatted correctly for ingest
 90        self.skip_reformat = skip_reformat
 91        """@private"""
 92
 93    @staticmethod
 94    def _reformat_for_type_consistency(ingest_metadata: list[dict]) -> list[dict]:
 95        """
 96        Reformats ingest metadata and finds headers where values are a mix of lists and non-lists.
 97
 98        If there is mix of these types of values, it converts the non-array to a one-item list. The updated metadata
 99        is then returned to be used for everything downstream
100        """
101        unique_headers = sorted({key for item in ingest_metadata for key in item.keys()})
102
103        headers_containing_mismatch = []
104        for header in unique_headers:
105            all_values_for_header = [r.get(header) for r in ingest_metadata]
106            # Find headers where some values are lists and some are not (while filtering out None values)
107            if any(isinstance(value, list) for value in all_values_for_header if value is not None) and not all(
108                    isinstance(value, list) for value in all_values_for_header if value is not None):
109                logging.info(
110                    f"Header {header} contains lists and non-list items. Will convert the non-list items into a list"
111                )
112                headers_containing_mismatch.append(header)
113
114        updated_metadata = []
115        for record in ingest_metadata:
116            new_record = {}
117            for header, value in record.items():
118                if header in headers_containing_mismatch:
119                    updated_value = [value] if not isinstance(value, list) else value
120                else:
121                    updated_value = value
122                new_record[header] = updated_value
123            updated_metadata.append(new_record)
124
125        return updated_metadata
126
127    def run(self) -> None:
128        """Run the batch ingestion process."""
129        logging.info(
130            f"Batching {len(self.ingest_metadata)} total rows into batches of {self.batch_size} for ingest")
131        total_batches = math.ceil(len(self.ingest_metadata) / self.batch_size)
132        for i in range(0, len(self.ingest_metadata), self.batch_size):
133            batch_number = i // self.batch_size + 1
134            logging.info(f"Starting ingest batch {batch_number} of {total_batches} into table {self.target_table_name}")
135            metrics_batch = self.ingest_metadata[i:i + self.batch_size]
136            if self.skip_reformat:
137                reformatted_batch = metrics_batch
138            else:
139                reformatted_batch = ReformatMetricsForIngest(
140                    ingest_metadata=metrics_batch,
141                    file_to_uuid_dict=self.file_to_uuid_dict,
142                    schema_info=self.schema_info
143                ).run()
144
145            if self.load_tag:
146                load_tag = self.load_tag
147            else:
148                load_tag = f"{self.dataset_id}.{self.target_table_name}"
149            # Start actual ingest
150            if reformatted_batch:
151                StartAndMonitorIngest(
152                    tdr=self.tdr,
153                    ingest_records=reformatted_batch,
154                    target_table_name=self.target_table_name,
155                    dataset_id=self.dataset_id,
156                    load_tag=load_tag,
157                    bulk_mode=self.bulk_mode,
158                    update_strategy=self.update_strategy,
159                    waiting_time_to_poll=self.waiting_time_to_poll
160                ).run()
161                logging.info(f"Completed batch ingest of {len(reformatted_batch)} rows")
162                if self.test_ingest:
163                    logging.info("First batch completed, exiting since test_ingest was used")
164                    sys.exit(0)
165            else:
166                logging.info("No rows to ingest in this batch after reformatting")
167        logging.info("Whole Ingest completed")
168
169
170class StartAndMonitorIngest:
171    """Class to start and monitor the ingestion of recordsinto a TDR dataset."""
172
173    def __init__(
174            self, tdr: TDR,
175            ingest_records: list[dict],
176            target_table_name: str,
177            dataset_id: str,
178            load_tag: str,
179            bulk_mode: bool,
180            update_strategy: str,
181            waiting_time_to_poll: int
182    ):
183        """
184        Initialize the StartAndMonitorIngest.
185
186        **Args:**
187        - tdr (`ops_utils.tdr_utils.tdr_api_utils.TDR`): An instance of the TDR class.
188        - ingest_records (list[dict]): The records to be ingested.
189        - target_table_name (str): The name of the target table.
190        - dataset_id (str): The ID of the dataset.
191        - load_tag (str): A tag to identify the load.
192        - bulk_mode (bool): Flag indicating if bulk mode should be used.
193        - update_strategy (str): The strategy for updating existing records.
194        - waiting_time_to_poll (int): The time to wait between polling for job status.
195        """
196        self.tdr = tdr
197        """@private"""
198        self.ingest_records = ingest_records
199        """@private"""
200        self.target_table_name = target_table_name
201        """@private"""
202        self.dataset_id = dataset_id
203        """@private"""
204        self.load_tag = load_tag
205        """@private"""
206        self.bulk_mode = bulk_mode
207        """@private"""
208        self.update_strategy = update_strategy
209        """@private"""
210        self.waiting_time_to_poll = waiting_time_to_poll
211        """@private"""
212
213    def _create_ingest_dataset_request(self) -> Any:
214        """
215        Create the ingestDataset request body.
216
217        Returns:
218            Any: The request body for ingesting the dataset.
219        """
220        # https://support.terra.bio/hc/en-us/articles/23460453585819-How-to-ingest-and-update-TDR-data-with-APIs
221        load_dict = {
222            "format": "array",
223            "records": self.ingest_records,
224            "table": self.target_table_name,
225            "resolve_existing_files": "true",
226            "updateStrategy": self.update_strategy,
227            "load_tag": self.load_tag,
228            "bulkMode": "true" if self.bulk_mode else "false"
229        }
230        return json.dumps(load_dict)  # dict -> json
231
232    def run(self) -> None:
233        """Run the ingestion process and monitor the job until completion."""
234        ingest_request = self._create_ingest_dataset_request()
235        logging.info(f"Starting ingest to {self.dataset_id}")
236        ingest_response = self.tdr.ingest_to_dataset(dataset_id=self.dataset_id, data=ingest_request).json()
237        MonitorTDRJob(
238            tdr=self.tdr,
239            job_id=ingest_response["id"],
240            check_interval=self.waiting_time_to_poll,
241            return_json=False
242        ).run()
243
244
245class ReformatMetricsForIngest:
246    """A class to reformat metrics for ingestion into TDR (Terra Data Repository)."""
247
248    def __init__(
249            self,
250            ingest_metadata: list[dict],
251            file_to_uuid_dict: Optional[dict] = None,
252            schema_info: Optional[dict] = None
253    ):
254        """
255        Initialize the ReformatMetricsForIngest class.
256
257        This class is used to reformat metrics for ingest.
258        Assumes input JSON will be in the following format for GCP:
259        ```
260        {
261            "file_name": blob.name,
262            "file_path": f"gs://{self.bucket_name}/{blob.name}",
263            "content_type": blob.content_type or guess_type(blob.name)[0] or "application/octet-stream",
264            "file_extension": os.path.splitext(blob.name)[1],
265            "size_in_bytes": blob.size,
266            "md5_hash": blob.md5_hash
267        }
268        ```
269
270        **Args:**
271        - ingest_metadata (list[dict]): The metadata to be ingested.
272        - file_to_uuid_dict (dict, optional): Only useful for self-hosted dataset. Can get from
273            create_file_uuid_dict_for_ingest_for_experimental_self_hosted_dataset. A dictionary mapping
274            source file paths to UUIDs. If used will make ingest much quicker since no ingest
275            or look up of file needed. Defaults to None.
276        - schema_info (dict, optional): Schema information for the tables. Defaults to None.
277        """
278        self.ingest_metadata = ingest_metadata
279        """@private"""
280        self.file_prefix = "gs://"
281        """@private"""
282        self.file_to_uuid_dict = file_to_uuid_dict
283        """@private"""
284        self.schema_info = schema_info
285        """@private"""
286
287    def _add_file_ref(self, file_details: dict) -> None:
288        """
289        Create file ref for ingest.
290
291        Args:
292            file_details (dict): The details of the file to be ingested.
293        """
294        file_details["file_ref"] = {
295            "sourcePath": file_details["path"],
296            "targetPath": self._format_relative_tdr_path(file_details["path"]),
297            "description": f"Ingest of {file_details['path']}",
298            "mimeType": file_details["content_type"]
299        }
300
301    @staticmethod
302    def _format_relative_tdr_path(cloud_path: str) -> str:
303        """
304        Format cloud path to TDR path.
305
306        Args:
307            cloud_path (str): The cloud path to be formatted.
308
309        Returns:
310            str: The formatted TDR path.
311        """
312        relative_path = "/".join(cloud_path.split("/")[3:])
313        return f"/{relative_path}"
314
315    def _check_and_format_file_path(self, column_value: str) -> tuple[Any, bool]:
316        """
317        Check if column value is a gs:// path and reformat to dict with ingest information.
318
319        If file_to_uuid_dict is
320        provided then it will add existing UUID. If file_to_uuid_dict provided and file not found then will warn and
321        return None.
322
323        Args:
324            column_value (str): The column value to be checked and formatted.
325
326        Returns:
327            tuple[Any, bool]: The formatted column value and a validity flag.
328        """
329        valid = True
330        if isinstance(column_value, str):
331            if column_value.startswith(self.file_prefix):
332                if self.file_to_uuid_dict:
333                    uuid = self.file_to_uuid_dict.get(column_value)
334                    if uuid:
335                        column_value = uuid
336                        return column_value, valid
337                    else:
338                        logging.warning(
339                            f"File {column_value} not found in file_to_uuid_dict, will attempt "
340                            "to ingest as regular file and not use UUID"
341                        )
342                source_dest_mapping = {
343                    "sourcePath": column_value,
344                    "targetPath": self._format_relative_tdr_path(column_value)
345                }
346                return source_dest_mapping, valid
347        return column_value, valid
348
349    def _validate_and_update_column_for_schema(self, column_name: str, column_value: Any) -> tuple[str, bool]:
350        """
351        Check if column matches what schema expects and attempt to update if not. Changes to string at the end.
352
353        Args:
354            column_name (str): The name of the column.
355            column_value (Any): The value of the column.
356
357        Returns:
358            tuple[str, bool]: The validated and updated column value and a validity flag.
359        """
360        valid = True
361        if self.schema_info:
362            if column_name in self.schema_info.keys():
363                expected_data_type = self.schema_info[column_name]["datatype"]
364                if expected_data_type == "string" and not isinstance(column_value, str):
365                    try:
366                        column_value = str(column_value)
367                    except ValueError:
368                        logging.warning(f"Column {column_name} with value {column_value} is not a string")
369                        valid = False
370                if expected_data_type in ["int64", "integer"] and not isinstance(column_value, int):
371                    try:
372                        column_value = int(column_value)
373                    except ValueError:
374                        logging.warning(f"Column {column_name} with value {column_value} is not an integer")
375                        valid = False
376                if expected_data_type == "float64" and not isinstance(column_value, float):
377                    try:
378                        column_value = float(column_value)
379                    except ValueError:
380                        logging.warning(f"Column {column_name} with value {column_value} is not a float")
381                        valid = False
382                if expected_data_type == "boolean" and not isinstance(column_value, bool):
383                    try:
384                        column_value = bool(column_value)
385                    except ValueError:
386                        logging.warning(f"Column {column_name} with value {column_value} is not a boolean")
387                        valid = False
388                if expected_data_type in ["datetime", "date", "time"] and not isinstance(column_value, datetime):
389                    try:
390                        column_value = parser.parse(column_value)
391                    except ValueError:
392                        logging.warning(f"Column {column_name} with value {column_value} is not a datetime")
393                        valid = False
394                if expected_data_type == "array" and not isinstance(column_value, list):
395                    valid = False
396                    logging.warning(f"Column {column_name} with value {column_value} is not a list")
397                if expected_data_type == "bytes" and not isinstance(column_value, bytes):
398                    valid = False
399                    logging.warning(f"Column {column_name} with value {column_value} is not bytes")
400                if expected_data_type == "fileref" and not column_value.startswith(self.file_prefix):
401                    valid = False
402                    logging.warning(f"Column {column_name} with value {column_value} is not a file path")
403        return str(column_value), valid
404
405    def _reformat_metric(self, row_dict: dict) -> Optional[dict]:
406        """
407        Reformat metric for ingest.
408
409        Args:
410            row_dict (dict): The row dictionary to be reformatted.
411
412        Returns:
413            Optional[dict]: The reformatted row dictionary or None if invalid.
414        """
415        reformatted_dict = {}
416        row_valid = True
417        for key, value in row_dict.items():
418            if value or value == 0:
419                if self.schema_info:
420                    value, valid = self._validate_and_update_column_for_schema(key, value)
421                    if not valid:
422                        row_valid = False
423                if isinstance(value, list):
424                    updated_value_list = []
425                    for item in value:
426                        update_value, valid = self._check_and_format_file_path(item)
427                        if not valid:
428                            row_valid = False
429                        updated_value_list.append(update_value)
430                    reformatted_dict[key] = updated_value_list
431                else:
432                    update_value, valid = self._check_and_format_file_path(value)
433                    reformatted_dict[key] = update_value
434                    if not valid:
435                        row_valid = False
436        reformatted_dict["last_modified_date"] = datetime.now(tz=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S")  # type: ignore[assignment] # noqa: E501
437        if row_valid:
438            return reformatted_dict
439        else:
440            logging.info(f"Row {json.dumps(row_dict, indent=4)} not valid and will not be included in ingest")
441            return None
442
443    def run(self) -> list[dict]:
444        """
445        Run the reformatting process for all metrics.
446
447        **Returns:**
448        - list[dict]: A list of reformatted metrics.
449        """
450        reformatted_metrics = []
451        for row_dict in self.ingest_metadata:
452            reformatted_row = self._reformat_metric(row_dict)
453            if reformatted_row:
454                reformatted_metrics.append(reformatted_row)
455        return reformatted_metrics
456
457
458class ConvertTerraTableInfoForIngest:
459    """Converts each row of table metadata into a dictionary that can be ingested into TDR."""
460
461    def __init__(
462            self,
463            table_metadata: list[dict],
464            columns_to_ignore: list[str] = [],
465            tdr_row_id: Optional[str] = None
466    ):
467        """
468        Initialize the ConvertTerraTableInfoForIngest class.
469
470        Input will look like this:
471        ```
472            [{
473              "attributes": {
474                "some_metric": 99.99,
475                "some_file_path": "gs://path/to/file",
476                "something_to_exclude": "exclude_me"
477              },
478              "entityType": "sample",
479              "name": "SM-MVVVV"
480            }]
481        ```
482        And be converted to this:
483        ```
484            [{
485              "sample_id": "SM-MVVVV",
486              "some_metric": 99.99,
487              "some_file_path": "gs://path/to/file"
488            }]
489        ```
490        **Args:**
491        - table_metadata (list[dict]): The metadata of the table to be converted.
492        - tdr_row_id (str): The row ID to be used in the TDR. Defaults to {entityType}_id.
493        - columns_to_ignore (list[str]): List of columns to ignore during conversion. Defaults to an empty list.
494        """
495        self.table_metadata = table_metadata
496        """@private"""
497        if table_metadata:
498            self.tdr_row_id = tdr_row_id if tdr_row_id else f'{table_metadata[0]["entityType"]}_id'
499            """@private"""
500        else:
501            # Won't be used if table_metadata is empty but will be set to empty string
502            self.tdr_row_id = ""
503            """@private"""
504        self.columns_to_ignore = columns_to_ignore
505        """@private"""
506
507    def run(self) -> list[dict]:
508        """
509        Convert the table metadata into a format suitable for TDR ingestion.
510
511        **Returns:**
512        - list[dict]: A list of dictionaries containing the converted table metadata.
513        """
514        return [
515            {
516                self.tdr_row_id: row["name"],
517                **{k: v for k, v in row["attributes"].items()
518                   # if columns_to_ignore is not provided or if the column is not in the columns_to_ignore list
519                   if k not in self.columns_to_ignore}
520            }
521            for row in self.table_metadata
522        ]
523
524
525class FilterAndBatchIngest:
526    """Filter and batch ingest process into TDR."""
527
528    def __init__(
529            self,
530            tdr: TDR,
531            filter_existing_ids: bool,
532            unique_id_field: str,
533            table_name: str,
534            ingest_metadata: list[dict],
535            dataset_id: str,
536            ingest_waiting_time_poll: int,
537            ingest_batch_size: int,
538            bulk_mode: bool,
539            update_strategy: str,
540            load_tag: str,
541            test_ingest: bool = False,
542            file_to_uuid_dict: Optional[dict] = None,
543            schema_info: Optional[dict] = None,
544            skip_reformat: bool = False
545    ):
546        """
547        Initialize the FilterAndBatchIngest class.
548
549        **Args:**
550        - tdr (`ops_utils.tdr_utils.tdr_api_utils.TDR`): An instance of the TDR class.
551        - filter_existing_ids (bool): Whether to filter out sample IDs that already exist in the dataset.
552        - unique_id_field (str): The unique ID field to filter on.
553        - table_name (str): The name of the table to ingest data into.
554        - ingest_metadata (list[dict]): The metadata to ingest.
555        - dataset_id (str): The ID of the dataset.
556        - ingest_waiting_time_poll (int): The waiting time to poll for ingest status.
557        - ingest_batch_size (int): The batch size for ingest.
558        - bulk_mode (bool): Whether to use bulk mode for ingest.
559        - update_strategy (str): The update strategy to use.
560        - load_tag (str): The load tag for ingest. Used to make future ingests of the same files go faster.
561        - test_ingest (bool, optional): Whether to run a test ingest. Defaults to False.
562        - file_to_uuid_dict (dict, optional): Only useful for self-hosted dataset. Can get from
563            create_file_uuid_dict_for_ingest_for_experimental_self_hosted_dataset. A dictionary mapping
564            source file paths to UUIDs. If used will make ingest much quicker since no ingest
565            or look up of file needed. Defaults to None.
566        - schema_info (dict, optional): Schema information for the tables.
567                Used to validate ingest metrics match. Defaults to None.
568        - skip_reformat (bool, optional): Whether to skip reformatting of metrics. Defaults to False.
569        """
570        self.tdr = tdr
571        """@private"""
572        self.filter_existing_ids = filter_existing_ids
573        """@private"""
574        self.unique_id_field = unique_id_field
575        """@private"""
576        self.table_name = table_name
577        """@private"""
578        self.ingest_metadata = ingest_metadata
579        """@private"""
580        self.dataset_id = dataset_id
581        """@private"""
582        self.ingest_waiting_time_poll = ingest_waiting_time_poll
583        """@private"""
584        self.ingest_batch_size = ingest_batch_size
585        """@private"""
586        self.bulk_mode = bulk_mode
587        """@private"""
588        self.update_strategy = update_strategy
589        """@private"""
590        self.load_tag = load_tag
591        """@private"""
592        self.test_ingest = test_ingest
593        """@private"""
594        self.file_to_uuid_dict = file_to_uuid_dict
595        """@private"""
596        self.schema_info = schema_info
597        """@private"""
598        self.skip_reformat = skip_reformat
599        """@private"""
600
601    def run(self) -> None:
602        """
603        Run the filter and batch ingest process.
604
605        This method filters out sample IDs that already exist in the dataset (if specified),
606        and then performs a batch ingest of the remaining metadata into the specified table.
607        """
608        if self.filter_existing_ids:
609            # Filter out sample ids that are already in the dataset
610            filtered_metrics = FilterOutSampleIdsAlreadyInDataset(
611                ingest_metrics=self.ingest_metadata,
612                dataset_id=self.dataset_id,
613                tdr=self.tdr,
614                target_table_name=self.table_name,
615                filter_entity_id=self.unique_id_field
616            ).run()
617        else:
618            filtered_metrics = self.ingest_metadata
619        # If there are metrics to ingest then ingest them
620        if filtered_metrics:
621            # Batch ingest of table to table within dataset
622            logging.info(f"Starting ingest into {self.table_name} in dataset {self.dataset_id}")
623            BatchIngest(
624                ingest_metadata=filtered_metrics,
625                tdr=self.tdr,
626                target_table_name=self.table_name,
627                dataset_id=self.dataset_id,
628                batch_size=self.ingest_batch_size,
629                bulk_mode=self.bulk_mode,
630                update_strategy=self.update_strategy,
631                waiting_time_to_poll=self.ingest_waiting_time_poll,
632                test_ingest=self.test_ingest,
633                load_tag=self.load_tag,
634                file_to_uuid_dict=self.file_to_uuid_dict,
635                schema_info=self.schema_info,
636                skip_reformat=self.skip_reformat
637            ).run()
638
639
640class GetPermissionsForWorkspaceIngest:
641    """Obtain permissions necessary for workspace ingest."""
642
643    def __init__(self, terra_workspace: TerraWorkspace, dataset_info: dict, added_to_auth_domain: bool = False):
644        """
645        Initialize the GetPermissionsForWorkspaceIngest class.
646
647        **Args:**
648        - terra_workspace (`ops_utils.terra_util.TerraWorkspace`): Instance of the TerraWorkspace class.
649        - dataset_info (dict): Information about the dataset.
650        - added_to_auth_domain (bool, optional): Flag indicating if the SA account
651                has been added to the auth domain. Defaults to `False`.
652        """
653        self.terra_workspace = terra_workspace
654        """@private"""
655        self.dataset_info = dataset_info
656        """@private"""
657        self.added_to_auth_domain = added_to_auth_domain
658        """@private"""
659
660    def run(self) -> None:
661        """
662        Ensure the dataset SA account has the necessary permissions on the Terra workspace.
663
664        This method updates the user ACL to make the SA account a reader on the Terra workspace.
665        It also checks if the workspace has an authorization domain, and logs the
666        necessary steps to add the SA account to the auth domain.
667        """
668        # Ensure dataset SA account is reader on Terra workspace.
669        tdr_sa_account = self.dataset_info["ingestServiceAccount"]
670        self.terra_workspace.update_user_acl(email=tdr_sa_account, access_level="READER")
671
672        # Check if workspace has auth domain
673        workspace_info = self.terra_workspace.get_workspace_info().json()
674        auth_domain_list = workspace_info["workspace"]["authorizationDomain"]
675        # Attempt to add tdr_sa_account to auth domain
676        if auth_domain_list:
677            for auth_domain_dict in auth_domain_list:
678                auth_domain = auth_domain_dict["membersGroupName"]
679                logging.info(f"TDR SA account {tdr_sa_account} needs to be added to auth domain group {auth_domain}")
680            if self.added_to_auth_domain:
681                logging.info("added_to_auth_domain has been set to true so assuming account has already been added")
682            else:
683                logging.info(
684                    f"Please add TDR SA account {tdr_sa_account} to auth domain group(s) to allow  "
685                    "access to workspace and then rerun with added_to_auth_domain=True"
686                )
687                sys.exit(0)