def _GetGcsUris()

in utils/import_conversations_v2.py [0:0]

• 15 lines of code
• 5 McCabe index (conditional complexity)

def _GetGcsUris(bucket, project_id, impersonated_service_account,
                folder_name=None, uri=True):
  """Returns a list of GCS uris for files in a bucket.
  
  Args:
      bucket: The GCS bucket.
      project_id: The project ID (not number) to use.
      impersonated_service_account: The service account to impersonate.
      folder_name: Folder path if the file is inside a nested path inside bucket
      uri: Whether to return gcs uri or not
  Returns:
      The GCS uris or file name.
  """
  uris = []
  storage_client = storage.Client(
      project=project_id,
      credentials=_GetClientCredentials(impersonated_service_account))
  blobs = storage_client.list_blobs(bucket, prefix=folder_name)
  for blob in blobs:
    # Blobs ending in slashes are actually directory paths.
    if not blob.name.endswith('/'):
      # Redaction Error: 0.5MB size
      if blob.size <= 1e7:
        if uri:
          uris.append(_GetGcsUri(bucket, blob.name))
        else:
          uris.append(blob.name)
  return uris