def gcs_read_file()

in python/activation/main.py [0:0]

• 12 lines of code
• 6 McCabe index (conditional complexity)

def gcs_read_file(project_id, gcs_path):
  """
  Reads a file from Google Cloud Storage (GCS).

  Args:
    project_id: The ID of the Google Cloud project that contains the GCS bucket.
    gcs_path: The path to the file in GCS, in the format "gs://bucket_name/object_name".

  Returns:
    The contents of the file as a string.

  Raises:
    ValueError: If the GCS path is invalid.
    IOError: If an error occurs while reading the file.
  """
  # Validate the GCS path.
  if not gcs_path.startswith("gs://"):
    raise ValueError("Invalid GCS path: {}".format(gcs_path))

  # Extract the bucket name and object name from the GCS path.
  matches = re.match("gs://(.*?)/(.*)", gcs_path)
  if not matches:
    raise ValueError("Invalid GCS path: {}".format(gcs_path))
  bucket_name, blob_name = matches.groups()

  # Create a storage client.
  storage_client = storage.Client(project=project_id)
  # Get a reference to the bucket and blob.
  bucket = storage_client.bucket(bucket_name)
  blob = bucket.blob(blob_name)
  # Open the blob for reading.
  with blob.open("r") as f:
    return f.read()