如何使用 Google Cloud API 获取给定存储桶中的文件夹列表
How to get list of folders in a given bucket using Google Cloud API
我想使用 Google 云存储 API.
获取给定 Google 云存储桶或文件夹中的所有文件夹
例如，如果 gs://abc/xyz 包含三个文件夹 gs://abc/xyz/x1、gs://abc/xyz/x2 和 gs://abc/xyz/x3。 API 应该 return gs://abc/xyz 中的所有三个文件夹。
使用gsutil
可以轻松完成
gsutil ls gs://abc/xyz
但我需要使用 python 和 Google 云存储 API。
您可以使用 Python GCS API 客户端库。有关文档和下载的相关链接，请参阅 Samples and Libraries for Google Cloud Storage 文档页面。
对于您的情况，首先我想指出您混淆了“桶”一词。我建议阅读文档的 Key Terms 页。你说的是对象名称前缀。
您可以从 list-objects.py sample on GitHub. Looking at the list 参考页开始，您需要通过 bucket=abc、prefix=xyz/ 和 delimiter=/。
我还需要简单地列出一个桶的内容。理想情况下，我想要类似于 tf.gfile 提供的东西。 tf.gfile 支持确定条目是文件还是目录。
我尝试了上面@jterrace 提供的各种链接，但我的结果并不是最佳的。话虽如此，值得展示结果。
给定一个混合了 "directories" 和 "files" 的存储桶，很难在 "filesystem" 中导航以找到感兴趣的项目。我在代码中提供了一些注释
关于上面引用的代码是如何工作的。
在任何一种情况下，我都在使用带有笔记本凭据的数据实验室笔记本。鉴于结果，我将需要使用字符串解析来确定特定目录中的文件。如果有人知道如何扩展这些方法或其他方法来解析类似于tf.gfile的目录，请回复。
方法一
import sys
import json
import argparse
import googleapiclient.discovery
BUCKET = 'bucket-sounds'
def create_service():
return googleapiclient.discovery.build('storage', 'v1')
def list_bucket(bucket):
"""Returns a list of metadata of the objects within the given bucket."""
service = create_service()
# Create a request to objects.list to retrieve a list of objects.
fields_to_return = 'nextPageToken,items(name,size,contentType,metadata(my-key))'
#req = service.objects().list(bucket=bucket, fields=fields_to_return) # returns everything
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound') # returns everything. UrbanSound is top dir in bucket
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/FREE') # returns the file FREESOUNDCREDITS.TXT
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/FREESOUNDCREDITS.txt', delimiter='/') # same as above
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/data/dog_bark', delimiter='/') # returns nothing
req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/data/dog_bark/', delimiter='/') # returns files in dog_bark dir
all_objects = []
# If you have too many items to list in one request, list_next() will
# automatically handle paging with the pageToken.
while req:
resp = req.execute()
all_objects.extend(resp.get('items', []))
req = service.objects().list_next(req, resp)
return all_objects
# usage
print(json.dumps(list_bucket(BUCKET), indent=2))
这会生成如下结果：
[
{
"contentType": "text/csv",
"name": "UrbanSound/data/dog_bark/100032.csv",
"size": "29"
},
{
"contentType": "application/json",
"name": "UrbanSound/data/dog_bark/100032.json",
"size": "1858"
} stuff snipped]
方法二
import re
import sys
from google.cloud import storage
BUCKET = 'bucket-sounds'
# Create a Cloud Storage client.
gcs = storage.Client()
# Get the bucket that the file will be uploaded to.
bucket = gcs.get_bucket(BUCKET)
def my_list_bucket(bucket_name, limit=sys.maxsize):
a_bucket = gcs.lookup_bucket(bucket_name)
bucket_iterator = a_bucket.list_blobs()
for resource in bucket_iterator:
print(resource.name)
limit = limit - 1
if limit <= 0:
break
my_list_bucket(BUCKET, limit=5)
这会生成这样的输出。
UrbanSound/FREESOUNDCREDITS.txt
UrbanSound/UrbanSound_README.txt
UrbanSound/data/air_conditioner/100852.csv
UrbanSound/data/air_conditioner/100852.json
UrbanSound/data/air_conditioner/100852.mp3
这是对此回答线程的更新：
from google.cloud import storage
# Instantiates a client
storage_client = storage.Client()
# Get GCS bucket
bucket = storage_client.get_bucket(bucket_name)
# Get blobs in bucket (including all subdirectories)
blobs_all = list(bucket.list_blobs())
# Get blobs in specific subirectory
blobs_specific = list(bucket.list_blobs(prefix='path/to/subfolder/'))
要获取存储桶中的文件夹列表，您可以使用以下代码片段：
import googleapiclient.discovery
def list_sub_directories(bucket_name, prefix):
"""Returns a list of sub-directories within the given bucket."""
service = googleapiclient.discovery.build('storage', 'v1')
req = service.objects().list(bucket=bucket_name, prefix=prefix, delimiter='/')
res = req.execute()
return res['prefixes']
# For the example (gs://abc/xyz), bucket_name is 'abc' and the prefix would be 'xyz/'
print(list_sub_directories(bucket_name='abc', prefix='xyz/'))
# sudo pip3 install --upgrade google-cloud-storage
from google.cloud import storage
os.environ["GOOGLE_APPLICATION_CREDENTIALS"]= "./key.json"
storage_client = storage.Client()
bucket = storage_client.get_bucket("my-bucket")
blobs = list(bucket.list_blobs(prefix='dir/'))
print (blobs)
这个问题是关于在 bucket/folder 中列出 文件夹 。 None 的建议对我有用，在试用 google.cloud.storage SDK 后，我怀疑（截至 2019 年 11 月）无法列出存储桶中任何路径的子目录。 REST API 是可能的，所以我写了这个小包装器...
from google.api_core import page_iterator
from google.cloud import storage
def _item_to_value(iterator, item):
return item
def list_directories(bucket_name, prefix):
if prefix and not prefix.endswith('/'):
prefix += '/'
extra_params = {
"projection": "noAcl",
"prefix": prefix,
"delimiter": '/'
}
gcs = storage.Client()
path = "/b/" + bucket_name + "/o"
iterator = page_iterator.HTTPIterator(
client=gcs,
api_request=gcs._connection.api_request,
path=path,
items_key='prefixes',
item_to_value=_item_to_value,
extra_params=extra_params,
)
return [x for x in iterator]
例如，如果您 my-bucket 包含：
- 狗叫声
- 数据集
- v1
- v2
然后调用 list_directories('my-bucket', 'dog-bark/datasets') 将 return:
['dog-bark/datasets/v1', 'dog-bark/datasets/v2']
#python notebook
ret_folders = !gsutil ls $path_possible_with_regex | grep -e "/$"
ret_folders_no_subdir = [x for x in srr_folders if x.split("/")[-2] != "SUBDIR”]
您可以将条件编辑为适合您的任何内容。就我而言，我只想要更深的“文件夹”。对于保存级别文件夹，您可以替换为
x.split("/")[-2] == "SUBDIR”
我遇到了同样的问题，并设法通过使用 list_blobs 描述的标准来完成它 here:
from google.cloud import storage
storage_client = storage.Client()
# Note: Client.list_blobs requires at least package version 1.17.0.
blobs = storage_client.list_blobs(
bucket_name, prefix=prefix, delimiter=delimiter
)
print("Blobs:")
for blob in blobs:
print(blob.name)
if delimiter:
print("Prefixes:")
for prefix in blobs.prefixes:
print(prefix)
然而，这只在我阅读 AntPhitlok 后才对我有用一个分隔符。
因此，在 'Blobs:' 部分下，我们只会获取文件名，而不是文件夹（如果存在于前缀文件夹下）。
所有 sub-directories 都将列在 'Prefixes:' 部分下。
重要的是要注意 blobs 实际上是一个迭代器，所以为了得到 sub-directories，我们必须“打开”它。因此，从我们的代码中省略 'Blobs:' 部分，将导致 blobs.prefixes
中的空 set()
编辑：
用法示例 - 假设我有一个名为 buck 的存储桶，其中有一个名为 dir 的目录。在 dir 我有另一个名为 subdir.
的目录
为了列出dir里面的目录，我可以使用：
from google.cloud import storage
storage_client = storage.Client()
blobs = storage_client.list_blobs('buck', prefix='dir/', delimiter='/')
print("Blobs:")
for blob in blobs:
print(blob.name)
if delimiter:
print("Prefixes:")
for prefix in blobs.prefixes:
print(prefix)
*注意前缀末尾的 / 作为分隔符。
此调用将向我打印以下内容：
Prefixes:
subdir/
这是获取所有子文件夹的简单方法：
from google.cloud import storage
def get_subdirs(bucket_name, dir_name=None):
"""
List all subdirectories for a bucket or
a specific folder in a bucket. If `dir_name`
is left blank, it will list all directories in the bucket.
"""
client = storage.Client()
bucket = client.lookup_bucket(bucket_name)
all_folders = []
for resource in bucket.list_blobs(prefix=dir_name):
# filter for directories only
n = resource.name
if n.endswith("/"):
all_folders.append(n)
return all_folders
# Use as follows:
all_folders = get_subdirs("my-bucket")
这是一个简单的解决方案
from google.cloud import storage # !pip install --upgrade google-cloud-storage
import os
# set up your bucket
client = storage.Client()
storage_client = storage.Client.from_service_account_json('XXXXXXXX')
bucket = client.get_bucket('XXXXXXXX')
# get all the folder in folder "base_folder"
base_folder = 'model_testing'
blobs=list(bucket.list_blobs(prefix=base_folder))
folders = list(set([os.path.dirname(k.name) for k in blobs]))
print(*folders, sep = '\n')
如果您只想要所选文件夹中的文件夹
base_folder = base_folder.rstrip(os.sep) # needed to remove any slashes at the end of the string
one_out = list(set([base_folder+ os.sep.join(k.split(base_folder)[-1].split(os.sep)[:2]) for k in folders]))
print(*one_out, sep = '\n')
当然不用
list(set())
你可以使用 numpy
import numpy as np
np.unique()
1。访问您的客户端对象。
密码在哪里运行?
我在 Google 云平台 (GCP)
中（某处）
如果您从 GCP 内部访问 Google 云存储 (GCS)，例如 Google Kubernetes Engine (GKE)，您应该使用 工作负载身份 将您的 GKE 服务帐户配置为 GCS 服务帐户。 https://cloud.google.com/kubernetes-engine/docs/how-to/workload-identity
一旦你这样做，创建你的客户端就像
一样简单
import google.cloud.storage as gcs
client = gcs.Client()
在野外
如果您在其他地方：AWS、Azure、您的开发机器或 GCP 之外的其他地方，那么您需要在创建您下载的服务帐户密钥之间做出选择（它是一个带有加密的 json 文件其中包含私钥）或使用 工作负载身份联合 ，例如由 AWS、Azure 和“朋友”提供的。
假设您已决定将新的 GCS 服务帐户文件下载到 /secure/gcs.json。
PROJECT_NAME = "MY-GCP-PROJECT"
from google.oauth2.service_account import Credentials
import google.cloud.storage as gcs
client = gcs.Client(
project=PROJECT_NAME,
credentials=Credentials.from_service_account_file("/secure/gcs.json"),
)
2。向 GCS 发出列表文件夹请求
在 OP 中，我们正在尝试获取存储桶 abc 中路径 xyz 内的文件夹。请注意，与 Linux 不同，GCS 中的路径不以 / 开头，但是它们应该以 1 结尾。所以我们将寻找前缀为 xyz/ 的文件夹。那只是文件夹，而不是文件夹及其所有子文件夹。
BUCKET_NAME = "abc"
blobs = client.list_blobs(
BUCKET_NAME,
prefix="xyz/", # <- you need the trailing slash
delimiter="/",
max_results=1,
)
请注意我们只要求一个 blob。这不是一个错误：blob 就是文件本身——我们只对文件夹感兴趣。将 max_results 设置为零不起作用，请参见下文。
3。强制延迟加载...错误..加载！
这里的几个答案已经遍历了迭代器 blobs 中的每个元素，这可能有数百万次，但我们不需要那样做。也就是说，如果我们不遍历 any 元素，blobs 根本不会向 GCS 发出 api 请求。
next(blobs, ...) # Force blobs to load.
print(blobs.prefixes)
blobs 变量是一个最多包含一个元素的迭代器，但是，如果您的文件夹中（在其级别）没有文件，则元素可能为零。如果元素为零，则 next(blobs) 将引发 StopIteration.
第二个参数，省略号 ...，只是我选择的默认 return 值，应该没有 next 元素。我觉得这比 None 更具可读性，因为它向 reader 暗示这里正在发生一些值得注意的事情。毕竟，请求一个值只是在同一行丢弃它的代码确实具有潜在错误的所有特征，因此最好让我们的 reader 放心这是故意的。
最后，假设我们在aaa、bbb、ccc的xyz下有一个树结构，然后在ccc下我们有子子文件夹zzz。输出将是
{'xyz/aaa', 'xyz/bbb', 'xyz/ccc'}
请注意，根据 OP 的要求，我们看不到子文件夹 xyz/ccc/zzz。
如果有人不想经历 google-cloud-api 的学习曲线，您基本上可以使用 subprocess 模块来 运行 bash 命令：
import subprocess
out=subprocess.run(["gsutil","ls","path/to/some/folder/"], capture_output=True)
out_list = out.stdout.decode("utf-8").split("\n")
dir_list = [i for i in out_list if i.endswith("/")]
files_list = [i for i in out_list if i not in dir_list]
我想使用 Google 云存储 API.
获取给定 Google 云存储桶或文件夹中的所有文件夹例如，如果 gs://abc/xyz 包含三个文件夹 gs://abc/xyz/x1、gs://abc/xyz/x2 和 gs://abc/xyz/x3。 API 应该 return gs://abc/xyz 中的所有三个文件夹。
使用gsutil
gsutil ls gs://abc/xyz
但我需要使用 python 和 Google 云存储 API。
您可以使用 Python GCS API 客户端库。有关文档和下载的相关链接，请参阅 Samples and Libraries for Google Cloud Storage 文档页面。
对于您的情况，首先我想指出您混淆了“桶”一词。我建议阅读文档的 Key Terms 页。你说的是对象名称前缀。
您可以从 list-objects.py sample on GitHub. Looking at the list 参考页开始，您需要通过 bucket=abc、prefix=xyz/ 和 delimiter=/。
我还需要简单地列出一个桶的内容。理想情况下，我想要类似于 tf.gfile 提供的东西。 tf.gfile 支持确定条目是文件还是目录。
我尝试了上面@jterrace 提供的各种链接，但我的结果并不是最佳的。话虽如此，值得展示结果。
给定一个混合了 "directories" 和 "files" 的存储桶，很难在 "filesystem" 中导航以找到感兴趣的项目。我在代码中提供了一些注释 关于上面引用的代码是如何工作的。
在任何一种情况下，我都在使用带有笔记本凭据的数据实验室笔记本。鉴于结果，我将需要使用字符串解析来确定特定目录中的文件。如果有人知道如何扩展这些方法或其他方法来解析类似于tf.gfile的目录，请回复。
方法一
import sys
import json
import argparse
import googleapiclient.discovery
BUCKET = 'bucket-sounds'
def create_service():
return googleapiclient.discovery.build('storage', 'v1')
def list_bucket(bucket):
"""Returns a list of metadata of the objects within the given bucket."""
service = create_service()
# Create a request to objects.list to retrieve a list of objects.
fields_to_return = 'nextPageToken,items(name,size,contentType,metadata(my-key))'
#req = service.objects().list(bucket=bucket, fields=fields_to_return) # returns everything
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound') # returns everything. UrbanSound is top dir in bucket
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/FREE') # returns the file FREESOUNDCREDITS.TXT
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/FREESOUNDCREDITS.txt', delimiter='/') # same as above
#req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/data/dog_bark', delimiter='/') # returns nothing
req = service.objects().list(bucket=bucket, fields=fields_to_return, prefix='UrbanSound/data/dog_bark/', delimiter='/') # returns files in dog_bark dir
all_objects = []
# If you have too many items to list in one request, list_next() will
# automatically handle paging with the pageToken.
while req:
resp = req.execute()
all_objects.extend(resp.get('items', []))
req = service.objects().list_next(req, resp)
return all_objects
# usage
print(json.dumps(list_bucket(BUCKET), indent=2))
这会生成如下结果：
[
{
"contentType": "text/csv",
"name": "UrbanSound/data/dog_bark/100032.csv",
"size": "29"
},
{
"contentType": "application/json",
"name": "UrbanSound/data/dog_bark/100032.json",
"size": "1858"
} stuff snipped]
方法二
import re
import sys
from google.cloud import storage
BUCKET = 'bucket-sounds'
# Create a Cloud Storage client.
gcs = storage.Client()
# Get the bucket that the file will be uploaded to.
bucket = gcs.get_bucket(BUCKET)
def my_list_bucket(bucket_name, limit=sys.maxsize):
a_bucket = gcs.lookup_bucket(bucket_name)
bucket_iterator = a_bucket.list_blobs()
for resource in bucket_iterator:
print(resource.name)
limit = limit - 1
if limit <= 0:
break
my_list_bucket(BUCKET, limit=5)
这会生成这样的输出。
UrbanSound/FREESOUNDCREDITS.txt
UrbanSound/UrbanSound_README.txt
UrbanSound/data/air_conditioner/100852.csv
UrbanSound/data/air_conditioner/100852.json
UrbanSound/data/air_conditioner/100852.mp3
这是对此回答线程的更新：
from google.cloud import storage
# Instantiates a client
storage_client = storage.Client()
# Get GCS bucket
bucket = storage_client.get_bucket(bucket_name)
# Get blobs in bucket (including all subdirectories)
blobs_all = list(bucket.list_blobs())
# Get blobs in specific subirectory
blobs_specific = list(bucket.list_blobs(prefix='path/to/subfolder/'))
要获取存储桶中的文件夹列表，您可以使用以下代码片段：
import googleapiclient.discovery
def list_sub_directories(bucket_name, prefix):
"""Returns a list of sub-directories within the given bucket."""
service = googleapiclient.discovery.build('storage', 'v1')
req = service.objects().list(bucket=bucket_name, prefix=prefix, delimiter='/')
res = req.execute()
return res['prefixes']
# For the example (gs://abc/xyz), bucket_name is 'abc' and the prefix would be 'xyz/'
print(list_sub_directories(bucket_name='abc', prefix='xyz/'))
# sudo pip3 install --upgrade google-cloud-storage
from google.cloud import storage
os.environ["GOOGLE_APPLICATION_CREDENTIALS"]= "./key.json"
storage_client = storage.Client()
bucket = storage_client.get_bucket("my-bucket")
blobs = list(bucket.list_blobs(prefix='dir/'))
print (blobs)
这个问题是关于在 bucket/folder 中列出 文件夹 。 None 的建议对我有用，在试用 google.cloud.storage SDK 后，我怀疑（截至 2019 年 11 月）无法列出存储桶中任何路径的子目录。 REST API 是可能的，所以我写了这个小包装器...
from google.api_core import page_iterator
from google.cloud import storage
def _item_to_value(iterator, item):
return item
def list_directories(bucket_name, prefix):
if prefix and not prefix.endswith('/'):
prefix += '/'
extra_params = {
"projection": "noAcl",
"prefix": prefix,
"delimiter": '/'
}
gcs = storage.Client()
path = "/b/" + bucket_name + "/o"
iterator = page_iterator.HTTPIterator(
client=gcs,
api_request=gcs._connection.api_request,
path=path,
items_key='prefixes',
item_to_value=_item_to_value,
extra_params=extra_params,
)
return [x for x in iterator]
例如，如果您 my-bucket 包含：
- 狗叫声
- 数据集
- v1
- v2
- 数据集
然后调用 list_directories('my-bucket', 'dog-bark/datasets') 将 return:
['dog-bark/datasets/v1', 'dog-bark/datasets/v2']
#python notebook
ret_folders = !gsutil ls $path_possible_with_regex | grep -e "/$"
ret_folders_no_subdir = [x for x in srr_folders if x.split("/")[-2] != "SUBDIR”]
您可以将条件编辑为适合您的任何内容。就我而言，我只想要更深的“文件夹”。对于保存级别文件夹，您可以替换为
x.split("/")[-2] == "SUBDIR”
我遇到了同样的问题，并设法通过使用 list_blobs 描述的标准来完成它 here:
from google.cloud import storage
storage_client = storage.Client()
# Note: Client.list_blobs requires at least package version 1.17.0.
blobs = storage_client.list_blobs(
bucket_name, prefix=prefix, delimiter=delimiter
)
print("Blobs:")
for blob in blobs:
print(blob.name)
if delimiter:
print("Prefixes:")
for prefix in blobs.prefixes:
print(prefix)
然而，这只在我阅读 AntPhitlok
因此，在 'Blobs:' 部分下，我们只会获取文件名，而不是文件夹（如果存在于前缀文件夹下）。 所有 sub-directories 都将列在 'Prefixes:' 部分下。
重要的是要注意 blobs 实际上是一个迭代器，所以为了得到 sub-directories，我们必须“打开”它。因此，从我们的代码中省略 'Blobs:' 部分，将导致 blobs.prefixes
set()
编辑：
用法示例 - 假设我有一个名为 buck 的存储桶，其中有一个名为 dir 的目录。在 dir 我有另一个名为 subdir.
为了列出dir里面的目录，我可以使用：
from google.cloud import storage
storage_client = storage.Client()
blobs = storage_client.list_blobs('buck', prefix='dir/', delimiter='/')
print("Blobs:")
for blob in blobs:
print(blob.name)
if delimiter:
print("Prefixes:")
for prefix in blobs.prefixes:
print(prefix)
*注意前缀末尾的 / 作为分隔符。
此调用将向我打印以下内容：
Prefixes:
subdir/
这是获取所有子文件夹的简单方法：
from google.cloud import storage
def get_subdirs(bucket_name, dir_name=None):
"""
List all subdirectories for a bucket or
a specific folder in a bucket. If `dir_name`
is left blank, it will list all directories in the bucket.
"""
client = storage.Client()
bucket = client.lookup_bucket(bucket_name)
all_folders = []
for resource in bucket.list_blobs(prefix=dir_name):
# filter for directories only
n = resource.name
if n.endswith("/"):
all_folders.append(n)
return all_folders
# Use as follows:
all_folders = get_subdirs("my-bucket")
这是一个简单的解决方案
from google.cloud import storage # !pip install --upgrade google-cloud-storage
import os
# set up your bucket
client = storage.Client()
storage_client = storage.Client.from_service_account_json('XXXXXXXX')
bucket = client.get_bucket('XXXXXXXX')
# get all the folder in folder "base_folder"
base_folder = 'model_testing'
blobs=list(bucket.list_blobs(prefix=base_folder))
folders = list(set([os.path.dirname(k.name) for k in blobs]))
print(*folders, sep = '\n')
如果您只想要所选文件夹中的文件夹
base_folder = base_folder.rstrip(os.sep) # needed to remove any slashes at the end of the string
one_out = list(set([base_folder+ os.sep.join(k.split(base_folder)[-1].split(os.sep)[:2]) for k in folders]))
print(*one_out, sep = '\n')
当然不用
list(set())
你可以使用 numpy
import numpy as np
np.unique()
1。访问您的客户端对象。
密码在哪里运行?
我在 Google 云平台 (GCP)
中（某处）如果您从 GCP 内部访问 Google 云存储 (GCS)，例如 Google Kubernetes Engine (GKE)，您应该使用 工作负载身份 将您的 GKE 服务帐户配置为 GCS 服务帐户。 https://cloud.google.com/kubernetes-engine/docs/how-to/workload-identity
一旦你这样做，创建你的客户端就像
一样简单import google.cloud.storage as gcs
client = gcs.Client()
在野外
如果您在其他地方：AWS、Azure、您的开发机器或 GCP 之外的其他地方，那么您需要在创建您下载的服务帐户密钥之间做出选择（它是一个带有加密的 json 文件其中包含私钥）或使用 工作负载身份联合 ，例如由 AWS、Azure 和“朋友”提供的。
假设您已决定将新的 GCS 服务帐户文件下载到 /secure/gcs.json。
PROJECT_NAME = "MY-GCP-PROJECT"
from google.oauth2.service_account import Credentials
import google.cloud.storage as gcs
client = gcs.Client(
project=PROJECT_NAME,
credentials=Credentials.from_service_account_file("/secure/gcs.json"),
)
2。向 GCS 发出列表文件夹请求
在 OP 中，我们正在尝试获取存储桶 abc 中路径 xyz 内的文件夹。请注意，与 Linux 不同，GCS 中的路径不以 / 开头，但是它们应该以 1 结尾。所以我们将寻找前缀为 xyz/ 的文件夹。那只是文件夹，而不是文件夹及其所有子文件夹。
BUCKET_NAME = "abc"
blobs = client.list_blobs(
BUCKET_NAME,
prefix="xyz/", # <- you need the trailing slash
delimiter="/",
max_results=1,
)
请注意我们只要求一个 blob。这不是一个错误：blob 就是文件本身——我们只对文件夹感兴趣。将 max_results 设置为零不起作用，请参见下文。
3。强制延迟加载...错误..加载！
这里的几个答案已经遍历了迭代器 blobs 中的每个元素，这可能有数百万次，但我们不需要那样做。也就是说，如果我们不遍历 any 元素，blobs 根本不会向 GCS 发出 api 请求。
next(blobs, ...) # Force blobs to load.
print(blobs.prefixes)
blobs 变量是一个最多包含一个元素的迭代器，但是，如果您的文件夹中（在其级别）没有文件，则元素可能为零。如果元素为零，则 next(blobs) 将引发 StopIteration.
第二个参数，省略号 ...，只是我选择的默认 return 值，应该没有 next 元素。我觉得这比 None 更具可读性，因为它向 reader 暗示这里正在发生一些值得注意的事情。毕竟，请求一个值只是在同一行丢弃它的代码确实具有潜在错误的所有特征，因此最好让我们的 reader 放心这是故意的。
最后，假设我们在aaa、bbb、ccc的xyz下有一个树结构，然后在ccc下我们有子子文件夹zzz。输出将是
{'xyz/aaa', 'xyz/bbb', 'xyz/ccc'}
请注意，根据 OP 的要求，我们看不到子文件夹 xyz/ccc/zzz。
如果有人不想经历 google-cloud-api 的学习曲线，您基本上可以使用 subprocess 模块来 运行 bash 命令：
import subprocess
out=subprocess.run(["gsutil","ls","path/to/some/folder/"], capture_output=True)
out_list = out.stdout.decode("utf-8").split("\n")
dir_list = [i for i in out_list if i.endswith("/")]
files_list = [i for i in out_list if i not in dir_list]