ops_utils.gcp_utils

  1"""Module for GCP utilities."""
  2import os
  3import logging
  4import io
  5import json
  6import hashlib
  7import base64
  8import subprocess
  9
 10from humanfriendly import format_size, parse_size
 11from mimetypes import guess_type
 12from typing import Optional, Any
 13from google.cloud.storage.blob import Blob
 14from google.oauth2 import service_account
 15from google.cloud import storage
 16from google.auth import default
 17
 18from .vars import ARG_DEFAULTS
 19from .thread_pool_executor_util import MultiThreadedJobs
 20
 21MOVE = "move"
 22"""Variable to be used for the "action" when files should be moved."""
 23COPY = "copy"
 24"""Variable to be used for the "action" when files should be copied."""
 25MD5_HEX = "hex"
 26"""Variable to be used when the generated `md5` should be of the `hex` type."""
 27MD5_BASE64 = "base64"
 28"""Variable to be used when the generated `md5` should be of the `base64` type."""
 29
 30
 31class GCPCloudFunctions:
 32    """Class to handle GCP Cloud Functions."""
 33
 34    def __init__(
 35            self,
 36            project: Optional[str] = None,
 37            service_account_json: Optional[str] = None
 38    ) -> None:
 39        """
 40        Initialize the GCPCloudFunctions class.
 41
 42        Authenticates using service account JSON if provided or default credentials,
 43        and sets up the Storage Client.
 44
 45        Args:
 46            project: Optional[str] = None
 47                The GCP project ID. If not provided, will use project from service account or default.
 48            service_account_json: Optional[str] = None
 49                Path to service account JSON key file. If provided, will use these credentials.
 50        """
 51        # Initialize credentials and project
 52        credentials = None
 53        default_project = None
 54
 55        if service_account_json:
 56            credentials = service_account.Credentials.from_service_account_file(service_account_json)
 57            # Extract project from service account if not specified
 58            if not project:
 59                with open(service_account_json, 'r') as f:
 60                    sa_info = json.load(f)
 61                    project = sa_info.get('project_id')
 62        else:
 63            # Use default credentials
 64            credentials, default_project = default()
 65
 66        # Set project if not already set
 67        if not project:
 68            project = default_project
 69
 70        self.client = storage.Client(credentials=credentials, project=project)
 71        """@private"""
 72
 73    @staticmethod
 74    def _process_cloud_path(cloud_path: str) -> dict:
 75        """
 76        Process a GCS cloud path into its components.
 77
 78        Args:
 79            cloud_path (str): The GCS cloud path.
 80
 81        Returns:
 82            dict: A dictionary containing the platform prefix, bucket name, and blob URL.
 83        """
 84        platform_prefix, remaining_url = str.split(str(cloud_path), sep="//", maxsplit=1)
 85        bucket_name = str.split(remaining_url, sep="/")[0]
 86        blob_name = "/".join(str.split(remaining_url, sep="/")[1:])
 87        path_components = {
 88            "platform_prefix": platform_prefix,
 89            "bucket": bucket_name,
 90            "blob_url": blob_name
 91        }
 92        return path_components
 93
 94    def load_blob_from_full_path(self, full_path: str) -> Blob:
 95        """
 96        Load a GCS blob object from a full GCS path.
 97
 98        **Args:**
 99        - full_path (str): The full GCS path.
100
101        **Returns:**
102        - google.cloud.storage.blob.Blob: The GCS blob object.
103        """
104        file_path_components = self._process_cloud_path(full_path)
105
106        # Specify the billing project
107        bucket = self.client.bucket(file_path_components["bucket"], user_project=self.client.project)
108        blob = bucket.blob(file_path_components["blob_url"])
109
110        # If blob exists in GCS reload it so metadata is there
111        if blob.exists():
112            blob.reload()
113        return blob
114
115    def check_file_exists(self, full_path: str) -> bool:
116        """
117        Check if a file exists in GCS.
118
119        **Args:**
120        - full_path (str): The full GCS path.
121
122        **Returns:**
123        - bool: `True` if the file exists, `False` otherwise.
124        """
125        blob = self.load_blob_from_full_path(full_path)
126        return blob.exists()
127
128    @staticmethod
129    def _create_bucket_contents_dict(bucket_name: str, blob: Any, file_name_only: bool) -> dict:
130        """
131        Create a dictionary containing file information.
132
133        Args:
134            bucket_name (str): The name of the GCS bucket.
135            blob (Any): The GCS blob object.
136            file_name_only (bool): Whether to return only the file list.
137
138        Returns:
139            dict: A dictionary containing file information.
140        """
141        if file_name_only:
142            return {
143                "path": f"gs://{bucket_name}/{blob.name}"
144            }
145        return {
146            "name": os.path.basename(blob.name),
147            "path": f"gs://{bucket_name}/{blob.name}",
148            "content_type": blob.content_type or guess_type(blob.name)[0] or "application/octet-stream",
149            "file_extension": os.path.splitext(blob.name)[1],
150            "size_in_bytes": blob.size,
151            "md5_hash": blob.md5_hash
152        }
153
154    @staticmethod
155    def _validate_include_blob(
156            blob: Any,
157            bucket_name: str,
158            file_extensions_to_ignore: list[str] = [],
159            file_strings_to_ignore: list[str] = [],
160            file_extensions_to_include: list[str] = [],
161            verbose: bool = False
162    ) -> bool:
163        """
164        Validate if a blob should be included based on its file extension.
165
166        Args:
167            file_extensions_to_include (list[str]): List of file extensions to include.
168            file_extensions_to_ignore (list[str]): List of file extensions to ignore.
169            file_strings_to_ignore (list[str]): List of file name substrings to ignore.
170            blob (Any): The GCS blob object.
171            verbose (bool): Whether to log files not being included.
172
173        Returns:
174            bool: True if the blob should be included, False otherwise.
175        """
176        file_path = f"gs://{bucket_name}/{blob.name}"
177        if file_extensions_to_ignore and file_path.endswith(tuple(file_extensions_to_ignore)):
178            if verbose:
179                logging.info(f"Skipping {file_path} as it has an extension to ignore")
180            return False
181        if file_extensions_to_include and not file_path.endswith(tuple(file_extensions_to_include)):
182            if verbose:
183                logging.info(f"Skipping {file_path} as it does not have an extension to include")
184            return False
185        if file_strings_to_ignore and any(file_string in file_path for file_string in file_strings_to_ignore):
186            if verbose:
187                logging.info(f"Skipping {file_path} as it has a string to ignore")
188            return False
189        return True
190
191    def list_bucket_contents(
192            self,
193            bucket_name: str,
194            prefix: Optional[str] = None,
195            file_extensions_to_ignore: list[str] = [],
196            file_strings_to_ignore: list[str] = [],
197            file_extensions_to_include: list[str] = [],
198            file_name_only: bool = False
199    ) -> list[dict]:
200        """
201        List contents of a GCS bucket and return a list of dictionaries with file information.
202
203        **Args:**
204        - bucket_name (str): The name of the GCS bucket. If includes `gs://`, it will be removed.
205        - prefix (str, optional): The prefix to filter the blobs. Defaults to None.
206        - file_extensions_to_ignore (list[str], optional): List of file extensions to ignore. Defaults to [].
207        - file_strings_to_ignore (list[str], optional): List of file name substrings to ignore. Defaults to [].
208        - file_extensions_to_include (list[str], optional): List of file extensions to include. Defaults to [].
209        - file_name_only (bool, optional): Whether to return only the file list and no extra info. Defaults to `False`.
210
211        **Returns:**
212        - list[dict]: A list of dictionaries containing file information.
213        """
214        # If the bucket name starts with gs://, remove it
215        if bucket_name.startswith("gs://"):
216            bucket_name = bucket_name.split("/")[2].strip()
217
218        logging.info(f"Accessing bucket: {bucket_name} with project: {self.client.project}")
219
220        # Get the bucket object and set user_project for Requester Pays
221        bucket = self.client.bucket(bucket_name, user_project=self.client.project)
222
223        # List blobs within the bucket
224        blobs = bucket.list_blobs(prefix=prefix)
225        logging.info("Finished listing blobs. Processing files now.")
226
227        # Create a list of dictionaries containing file information
228        file_list = [
229            self._create_bucket_contents_dict(
230                blob=blob, bucket_name=bucket_name, file_name_only=file_name_only
231            )
232            for blob in blobs
233            if self._validate_include_blob(
234                blob=blob,
235                file_extensions_to_ignore=file_extensions_to_ignore,
236                file_strings_to_ignore=file_strings_to_ignore,
237                file_extensions_to_include=file_extensions_to_include,
238                bucket_name=bucket_name
239            ) and not blob.name.endswith("/")
240        ]
241        logging.info(f"Found {len(file_list)} files in bucket")
242        return file_list
243
244    def copy_cloud_file(self, src_cloud_path: str, full_destination_path: str, verbose: bool = False) -> None:
245        """
246        Copy a file from one GCS location to another.
247
248        **Args:**
249        - src_cloud_path (str): The source GCS path.
250        - full_destination_path (str): The destination GCS path.
251        - verbose (bool, optional): Whether to log progress. Defaults to `False`.
252        """
253        try:
254            src_blob = self.load_blob_from_full_path(src_cloud_path)
255            dest_blob = self.load_blob_from_full_path(full_destination_path)
256
257            # Use rewrite so no timeouts
258            rewrite_token = False
259
260            while True:
261                rewrite_token, bytes_rewritten, bytes_to_rewrite = dest_blob.rewrite(
262                    src_blob, token=rewrite_token
263                )
264                if verbose:
265                    logging.info(
266                        f"{full_destination_path}: Progress so far: {bytes_rewritten}/{bytes_to_rewrite} bytes."
267                    )
268                if not rewrite_token:
269                    break
270
271        except Exception as e:
272            logging.error(
273                f"Encountered the following error while attempting to copy file from '{src_cloud_path}' to "
274                f"'{full_destination_path}': {e}. If this is a retryable error, it will be re-attempted"
275            )
276            raise
277
278    def delete_cloud_file(self, full_cloud_path: str) -> None:
279        """
280        Delete a file from GCS.
281
282        **Args:**
283        - full_cloud_path (str): The GCS path of the file to delete.
284        """
285        blob = self.load_blob_from_full_path(full_cloud_path)
286        blob.delete()
287
288    def move_cloud_file(self, src_cloud_path: str, full_destination_path: str) -> None:
289        """
290        Move a file from one GCS location to another.
291
292        **Args:**
293        - src_cloud_path (str): The source GCS path.
294        - full_destination_path (str): The destination GCS path.
295        """
296        self.copy_cloud_file(src_cloud_path, full_destination_path)
297        self.delete_cloud_file(src_cloud_path)
298
299    def get_filesize(self, target_path: str) -> int:
300        """
301        Get the size of a file in GCS.
302
303        **Args:**
304        - target_path (str): The GCS path of the file.
305
306        **Returns:**
307        - int: The size of the file in bytes.
308        """
309        blob = self.load_blob_from_full_path(target_path)
310        return blob.size
311
312    def validate_files_are_same(self, src_cloud_path: str, dest_cloud_path: str) -> bool:
313        """
314        Validate if two cloud files (source and destination) are identical based on their MD5 hashes.
315
316        **Args:**
317        - src_cloud_path (str): The source GCS path.
318        - dest_cloud_path (str): The destination GCS path.
319
320        **Returns:**
321        - bool: `True` if the files are identical, `False` otherwise.
322        """
323        src_blob = self.load_blob_from_full_path(src_cloud_path)
324        dest_blob = self.load_blob_from_full_path(dest_cloud_path)
325
326        # If either blob is None or does not exist
327        if not src_blob or not dest_blob or not src_blob.exists() or not dest_blob.exists():
328            return False
329        # If the MD5 hashes exist
330        if src_blob.md5_hash and dest_blob.md5_hash:
331            # And are the same return True
332            if src_blob.md5_hash == dest_blob.md5_hash:
333                return True
334        else:
335            # If md5 do not exist (for larger files they may not) check size matches
336            if src_blob.size == dest_blob.size:
337                return True
338        # Otherwise, return False
339        return False
340
341    def delete_multiple_files(
342            self,
343            files_to_delete: list[str],
344            workers: int = ARG_DEFAULTS["multithread_workers"],  # type: ignore[assignment]
345            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
346            verbose: bool = False,
347            job_complete_for_logging: int = 500
348    ) -> None:
349        """
350        Delete multiple cloud files in parallel using multi-threading.
351
352        **Args:**
353        - files_to_delete (list[str]): List of GCS paths of the files to delete.
354        - workers (int, optional): Number of worker threads. Defaults to `10`.
355        - max_retries (int, optional): Maximum number of retries. Defaults to `5`.
356        - verbose (bool, optional): Whether to log each job's success. Defaults to `False`.
357        - job_complete_for_logging (int, optional): The number of jobs to complete before logging. Defaults to `500`.
358        """
359        list_of_jobs_args_list = [[file_path] for file_path in set(files_to_delete)]
360
361        MultiThreadedJobs().run_multi_threaded_job(
362            workers=workers,
363            function=self.delete_cloud_file,
364            list_of_jobs_args_list=list_of_jobs_args_list,
365            max_retries=max_retries,
366            fail_on_error=True,
367            verbose=verbose,
368            collect_output=False,
369            jobs_complete_for_logging=job_complete_for_logging
370        )
371
372    def _validate_file_pair(self, source_file: str, full_destination_path: str) -> dict:
373        """
374        Validate if source and destination files are identical.
375
376        **Args:**
377        - source_file (str): The source file path.
378        - full_destination_path (str): The destination file path.
379
380        **Returns:**
381            dict: The file dictionary of the files with a boolean indicating if they are identical.
382        """
383        if self.validate_files_are_same(source_file, full_destination_path):
384            identical = True
385        else:
386            identical = False
387        return {"source_file": source_file, "full_destination_path": full_destination_path, "identical": identical}
388
389    def loop_and_log_validation_files_multithreaded(
390            self,
391            files_to_validate: list[dict],
392            log_difference: bool,
393            workers: int = ARG_DEFAULTS["multithread_workers"],  # type: ignore[assignment]
394            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
395            job_complete_for_logging: int = 500
396    ) -> list[dict]:
397        """
398        Validate if multiple cloud files are identical based on their MD5 hashes using multithreading.
399
400        **Args:**
401        - files_to_validate (list[dict]): List of dictionaries containing source and destination file paths.
402        - log_difference (bool): Whether to log differences if files are not identical. Set `False` if you are running
403                                   this at the start of a copy/move operation to check if files are already copied.
404        - workers (int, optional): Number of worker threads. Defaults to `10`.
405        - max_retries (int, optional): Maximum number of retries for all jobs. Defaults to `5`.
406        - job_complete_for_logging (int, optional): The number of jobs to complete before logging. Defaults to `500`.
407
408        **Returns:**
409        - list[dict]: List of dictionaries containing files that are **not** identical.
410        """
411        logging.info(f"Validating if {len(files_to_validate)} files are identical")
412
413        # Prepare jobs: pass the necessary arguments to each validation
414        jobs = [(file_dict['source_file'], file_dict['full_destination_path']) for file_dict in files_to_validate]
415
416        # Use multithreaded job runner to validate the files
417        checked_files = MultiThreadedJobs().run_multi_threaded_job(
418            workers=workers,
419            function=self._validate_file_pair,
420            list_of_jobs_args_list=jobs,
421            collect_output=True,
422            max_retries=max_retries,
423            jobs_complete_for_logging=job_complete_for_logging
424        )
425        # If any files failed to load, raise an exception
426        if files_to_validate and None in checked_files:  # type: ignore[operator]
427            logging.error("Failed to validate all files, could not load some blobs")
428            raise Exception("Failed to validate all files")
429
430        # Get all files that are not identical
431        not_identical_files = [
432            file_dict
433            for file_dict in checked_files  # type: ignore[operator, union-attr]
434            if not file_dict['identical']
435        ]
436        if not_identical_files:
437            if log_difference:
438                for file_dict in not_identical_files:
439                    logging.warning(
440                        f"File {file_dict['source_file']} and {file_dict['full_destination_path']} are not identical"
441                    )
442            logging.info(f"Validation complete. {len(not_identical_files)} files are not identical.")
443        return not_identical_files
444
445    def multithread_copy_of_files_with_validation(
446            self,
447            files_to_copy: list[dict],
448            workers: int = ARG_DEFAULTS["multithread_workers"],  # type: ignore[assignment]
449            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
450            skip_check_if_already_copied: bool = False
451    ) -> None:
452        """
453        Copy multiple files in parallel with validation.
454
455        **Args:**
456        - files_to_copy (list[dict]): List of dictionaries containing source and destination file paths.
457                Dictionary should have keys `source_file` and `full_destination_path`
458        - workers (int): Number of worker threads. Defaults to `10`.
459        - max_retries (int): Maximum number of retries. Defaults to `5`
460        - skip_check_if_already_copied (bool, optional): Whether to skip checking
461                if files are already copied and start copying right away. Defaults to `False`.
462        """
463        if skip_check_if_already_copied:
464            logging.info("Skipping check if files are already copied")
465            updated_files_to_move = files_to_copy
466        else:
467            updated_files_to_move = self.loop_and_log_validation_files_multithreaded(
468                files_to_copy,
469                log_difference=False,
470                workers=workers,
471                max_retries=max_retries
472            )
473        # If all files are already copied, return
474        if not updated_files_to_move:
475            logging.info("All files are already copied")
476            return None
477        logging.info(f"Attempting to {COPY} {len(updated_files_to_move)} files")
478        self.move_or_copy_multiple_files(updated_files_to_move, COPY, workers, max_retries)
479        logging.info(f"Validating all {len(updated_files_to_move)} new files are identical to original")
480        # Validate that all files were copied successfully
481        files_not_moved_successfully = self.loop_and_log_validation_files_multithreaded(
482            files_to_copy,
483            workers=workers,
484            log_difference=True,
485            max_retries=max_retries
486        )
487        if files_not_moved_successfully:
488            logging.error(f"Failed to copy {len(files_not_moved_successfully)} files")
489            raise Exception("Failed to copy all files")
490        logging.info(f"Successfully copied {len(updated_files_to_move)} files")
491        return None
492
493    def move_or_copy_multiple_files(
494            self, files_to_move: list[dict],
495            action: str,
496            workers: int = ARG_DEFAULTS["multithread_workers"],  # type: ignore[assignment]
497            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
498            verbose: bool = False,
499            jobs_complete_for_logging: int = 500
500    ) -> None:
501        """
502        Move or copy multiple files in parallel.
503
504        **Args:**
505        - files_to_move (list[dict]): List of dictionaries containing source and destination file paths.
506        - action (str): The action to perform (should be one of `ops_utils.gcp_utils.MOVE`
507        or `ops_utils.gcp_utils.COPY`).
508        - workers (int): Number of worker threads. Defaults to `10`.
509        - max_retries (int): Maximum number of retries. Defaults to `5`.
510        - verbose (bool, optional): Whether to log each job's success. Defaults to `False`.
511        - jobs_complete_for_logging (int, optional): The number of jobs to complete before logging. Defaults to `500`.
512
513        **Raises:**
514        - ValueError: If the action is not one of `ops_utils.gcp_utils.MOVE` or `ops_utils.gcp_utils.COPY`.
515        """
516        if action == MOVE:
517            cloud_function = self.move_cloud_file
518        elif action == COPY:
519            cloud_function = self.copy_cloud_file
520        else:
521            raise ValueError("Must either select move or copy")
522
523        list_of_jobs_args_list = [
524            [
525                file_dict['source_file'], file_dict['full_destination_path']
526            ]
527            for file_dict in files_to_move
528        ]
529        MultiThreadedJobs().run_multi_threaded_job(
530            workers=workers,
531            function=cloud_function,
532            list_of_jobs_args_list=list_of_jobs_args_list,
533            max_retries=max_retries,
534            fail_on_error=True,
535            verbose=verbose,
536            collect_output=False,
537            jobs_complete_for_logging=jobs_complete_for_logging
538        )
539
540    def read_file(self, cloud_path: str, encoding: str = 'utf-8') -> str:
541        """
542        Read the content of a file from GCS.
543
544        **Args:**
545        - cloud_path (str): The GCS path of the file to read.
546        - encoding (str, optional): The encoding to use. Defaults to `utf-8`.
547
548        **Returns:**
549        - bytes: The content of the file as bytes.
550        """
551        blob = self.load_blob_from_full_path(cloud_path)
552        # Download the file content as bytes
553        content_bytes = blob.download_as_bytes()
554        # Convert bytes to string
555        content_str = content_bytes.decode(encoding)
556        return content_str
557
558    def upload_blob(self, destination_path: str, source_file: str, custom_metadata: Optional[dict] = None) -> None:
559        """
560        Upload a file to GCS.
561
562        **Args:**
563        - destination_path (str): The destination GCS path.
564        - source_file (str): The source file path.
565        - custom_metadata (dict, optional): A dictionary of custom metadata to attach to the blob. Defaults to None.
566        """
567        blob = self.load_blob_from_full_path(destination_path)
568        if custom_metadata:
569            blob.metadata = custom_metadata
570        blob.upload_from_filename(source_file)
571
572    def get_object_md5(
573        self,
574        file_path: str,
575        # https://jbrojbrojbro.medium.com/finding-the-optimal-download-size-with-gcs-259dc7f26ad2
576        chunk_size: int = parse_size("256 KB"),
577        logging_bytes: int = parse_size("1 GB"),
578        returned_md5_format: str = "hex"
579    ) -> str:
580        """
581        Calculate the MD5 checksum of a file in GCS.
582
583        **Args:**
584        - file_path (str): The GCS path of the file.
585        - chunk_size (int, optional): The size of each chunk to read. Defaults to `256 KB`.
586        - logging_bytes (int, optional): The number of bytes to read before logging progress. Defaults to `1 GB`.
587        - returned_md5_format (str, optional): The format of the MD5 checksum to return. Defaults to `hex`.
588                Options are `ops_utils.gcp_utils.MD5_HEX` or `ops_utils.gcp_utils.MD5_BASE64`.
589
590        **Returns:**
591        - str: The MD5 checksum of the file.
592
593        **Raises:**
594        - ValueError: If the `returned_md5_format` is not one of `ops_utils.gcp_utils.MD5_HEX`
595        or `ops_utils.gcp_utils.MD5_BASE64`
596        """
597        if returned_md5_format not in ["hex", "base64"]:
598            raise ValueError("returned_md5_format must be 'hex' or 'base64'")
599
600        blob = self.load_blob_from_full_path(file_path)
601
602        # Create an MD5 hash object
603        md5_hash = hashlib.md5()
604
605        blob_size_str = format_size(blob.size)
606        logging.info(f"Streaming {file_path} which is {blob_size_str}")
607        # Use a BytesIO stream to collect data in chunks and upload it
608        buffer = io.BytesIO()
609        total_bytes_streamed = 0
610        # Keep track of the last logged size for data logging
611        last_logged = 0
612
613        with blob.open("rb") as source_stream:
614            while True:
615                chunk = source_stream.read(chunk_size)
616                if not chunk:
617                    break
618                md5_hash.update(chunk)
619                buffer.write(chunk)
620                total_bytes_streamed += len(chunk)
621                # Log progress every 1 gb if verbose used
622                if total_bytes_streamed - last_logged >= logging_bytes:
623                    logging.info(f"Streamed {format_size(total_bytes_streamed)} / {blob_size_str} so far")
624                    last_logged = total_bytes_streamed
625
626        if returned_md5_format == "hex":
627            md5 = md5_hash.hexdigest()
628            logging.info(f"MD5 (hex) for {file_path}: {md5}")
629        elif returned_md5_format == "base64":
630            md5 = base64.b64encode(md5_hash.digest()).decode("utf-8")
631            logging.info(f"MD5 (base64) for {file_path}: {md5}")
632        return md5
633
634    def set_acl_public_read(self, cloud_path: str) -> None:
635        """
636        Set the file in the bucket to be publicly readable.
637
638        **Args:**
639        - cloud_path (str): The GCS path of the file to be set as public readable.
640        """
641        blob = self.load_blob_from_full_path(cloud_path)
642        blob.acl.all().grant_read()
643        blob.acl.save()
644
645    def set_acl_group_owner(self, cloud_path: str, group_email: str) -> None:
646        """
647        Set the file in the bucket to grant OWNER permission to a specific group.
648
649        **Args:**
650        - cloud_path (str): The GCS path of the file.
651        - group_email (str): The email of the group to grant OWNER permission
652        """
653        blob = self.load_blob_from_full_path(cloud_path)
654        blob.acl.group(group_email).grant_owner()
655        blob.acl.save()
656
657    def set_metadata_cache_control(self, cloud_path: str, cache_control: str) -> None:
658        """
659        Set Cache-Control metadata for a file.
660
661        **Args:**
662        - cloud_path (str): The GCS path of the file.
663        - cache_control (str): The Cache-Control metadata to set.
664        """
665        blob = self.load_blob_from_full_path(cloud_path)
666        blob.cache_control = cache_control
667        blob.patch()
668
669    def get_file_contents_of_most_recent_blob_in_bucket(self, bucket_name: str) -> Optional[tuple[str, str]]:
670        """
671        Get the most recent blob in the bucket.
672        
673        If the blob with the most recent timestamp doesn't have
674        any logging besides the basic "storage_byte_hours" logging, will continue to look at the next most
675        recent file until a log with useful information is encountered. This is useful when combing through
676        GCP activity logs for Terra workspace buckets.
677
678        **Args:**
679        - bucket_name (str): The GCS bucket name.
680
681        **Returns:**
682        - Optional tuple of the blob found and the file contents from the blob
683        """
684        blobs = sorted(
685            self.client.list_blobs(bucket_name), key=lambda blob: blob.updated, reverse=True
686        )
687        for blob in blobs:
688            # Download the file contents as a string
689            file_contents = blob.download_as_text()
690
691            # Check if the content matches the undesired format
692            lines = file_contents.splitlines()
693            if len(lines) > 1 and lines[0] == '"bucket","storage_byte_hours"':
694                logging.info(f"Skipping file {blob.name} as it matches the undesired format.")
695                continue
696
697            # If it doesn't match the undesired format, return its content
698            logging.info(f"Found valid file: {blob.name}")
699            return blob, file_contents
700
701        logging.info("No valid files found.")
702        return None
703
704    def write_to_gcp_file(self, cloud_path: str, file_contents: str) -> None:
705        """
706        Write content to a file in GCS.
707
708        **Args:**
709        - cloud_path (str): The GCS path of the file to write.
710        - file_contents (str): The content to write.
711        """
712        blob = self.load_blob_from_full_path(cloud_path)
713        blob.upload_from_string(file_contents)
714        logging.info(f"Successfully wrote content to {cloud_path}")
715
716    @staticmethod
717    def get_active_gcloud_account() -> str:
718        """
719        Get the active GCP email for the current account.
720
721        **Returns:**
722        - str: The active GCP account email.
723        """
724        result = subprocess.run(
725            args=["gcloud", "auth", "list", "--filter=status:ACTIVE", "--format=value(account)"],
726            capture_output=True,
727            text=True,
728            check=True
729        )
730        return result.stdout.strip()
731
732