我想使用Google Cloud Storage API获取给定Google Cloud bucket或文件夹中的所有文件夹。
例如，如果gs://abc/xyz包含三个文件夹gs://abc/xyz/x1、gs://abc/xyz/x2和gs://abc/xyz/x3。该API应返回gs://abc/xyz中所有三个文件夹。
可以轻松地使用gsutil完成此操作。
gsutil ls gs://abc/xyz
但我需要使用Python和Google Cloud Storage API来完成它。
这个问题是关于列举存储桶(文件夹)内的子文件夹。针对所有建议都没有奏效，经过尝试使用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 的内容:
list_directories('my-bucket', 'dog-bark/datasets') 将返回:
['dog-bark/datasets/v1', 'dog-bark/datasets/v2']._connection。使用list_blobs(..)会更容易/更安全的方法。 - Robino''前缀列出存储桶的根目录 :) 除此之外，这对我来说完美地起作用了，感谢您的发布！ - Yet Another User这里是对回答帖子的更新：
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/'))
bucket=abc、prefix=xyz/和delimiter=/。prefix和delimiter调用objects().list()时，我们会得到一组匹配物体和匹配前缀的列表。如@jterrace所回答的那样，如果我们在prefix=abc/xyz和delimiter=/中传递，则会获取所有名称以abc/xyz开头的对象以及前缀，这些前缀可以被逻辑上视为子文件夹。 - Shamshad Alamgs://abc/xyz，那么存储桶将是abc。如果您还在前缀中传递存储桶名称，则可能无法匹配任何内容，而且肯定不是您想要的。 - Robino获取存储桶中文件夹列表，您可以使用下面的代码片段：
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/'))
google.cloud.storage API... - Robino代码在哪里运行？
如果您正在从GCP内部访问Google Cloud Storage（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()
如果你不在GCP内，比如在AWS、Azure、本地开发环境或其它地方，那么需要选择以下两种方式之一：要么创建一个服务账号并下载该账号的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"),
)
在这个操作中，我们试图获取存储桶abc中路径xyz内的文件夹列表。请注意，GCS中的路径不像Linux那样以/开头，但是它们应该以一个/结束。因此，我们将查找带有前缀xyz/ 的文件夹。也就是说，只是文件夹，而不是包含所有子文件夹的文件夹。
BUCKET_NAME = "abc"
blobs = client.list_blobs(
BUCKET_NAME,
prefix="xyz/", # <- you need the trailing slash
delimiter="/",
max_results=1,
)
注意我们只要求一个 blob，这不是一个错误：blobs 是文件本身，我们只对文件夹感兴趣。将 max_results 设置为零并不起作用，详见下文。
这里的一些回答遍历了迭代器中的每个元素 blobs，可能会遍历成千上万个，但我们不需要这样做。话虽如此，如果我们没有遍历任何元素，则 blobs 不会进行向 GCS 发出 API 请求的操作。
next(blobs, ...) # Force blobs to load.
print(blobs.prefixes)
blobs变量是一个迭代器，最多包含一个元素，但是如果您的文件夹（在其级别上）没有文件，则可能没有元素。如果没有元素，则next(blobs)将引发StopIteration异常。...，只是我选择的默认返回值，如果没有下一个元素。我觉得这比None更可读，因为它向读者暗示着这里发生了值得注意的事情。毕竟，请求值后立即在同一行中丢弃它的代码具有潜在错误的所有标志，因此让我们向读者保证这是有意为之的是很好的。xyz下有一个树形结构，即aaa、bbb、ccc，然后在ccc下面有子文件夹zzz。输出将会是：{'xyz/aaa', 'xyz/bbb', 'xyz/ccc'}
xyz/ccc/zzz。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
我遇到了同样的问题，并通过使用此处描述的标准list_blobs方法解决了它:
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：”部分下，如果存在于前缀文件夹下，则我们将仅获取文件名，而不是文件夹。 所有子目录都将列在“前缀：”部分下。
需要注意的是，blobs实际上是一个迭代器，因此为了获得子目录，我们必须“打开”它。因此，如果在我们的代码中省略“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/
prefix 开头的文件和文件夹。当您只想列出特定目录中的文件和文件夹时，它非常有用。重要的是，前缀应以“/”结尾。然而，分隔符有助于在该目录中分隔文件和文件夹。正如我所写的那样，我已经使用“/”作为分隔符。 - PeNpeLfrom 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())
import numpy as np
np.unique()
max_results=1，像这样 blobs=list(bucket.list_blobs(max_results=1, prefix=base_folder))？ - Phillip Maire跟随https://stackoverflow.com/users/2350164/yet-another-user的回答，我已经使用"标准"谷歌代替HTTPIterator创建了相同的函数。 假设我们有一个名为'bucket_name'的Bucket和一个名为'sub_folder_name'的子文件夹。
from google.api_core import page_iterator
from google.cloud import storage
storage_client = storage.Client(project = PROJECT_NAME)
def get_folders_list(storage_client, bucket_or_name, prefix = ''):
"""
the function returns the list of folders within a bucket or its subdirectory
:param storage_client: the GCS client
:param bucket_or_name: the name of the bucket
:param prefix: the prefix if you want subdirectory
:return: list of folders
"""
if prefix and not prefix.endswith('/'):
prefix += '/'
blobs = storage_client.list_blobs(
bucket_or_name=bucket_or_name,
prefix=prefix,
delimiter="/",
# max_results=1
)
next(blobs, ...)
return list(blobs.prefixes)
您可以使用以下两个示例来针对存储桶或其子目录之一：
get_folders_list(storage_client = storage_client, bucket_or_name =
'bucket_name')
get_folders_list(storage_client = storage_client, bucket_or_name = 'bucket_name', prefix = 'sub_folder_name')
您可以使用Python云存储库和一行代码来获取存储桶内深度为N的所有唯一前缀，例如当N=2时：
set(["/".join(blob.name.split('/',maxsplit=2)[0:2]) for blob in client.list_blobs(BUCKET_NAME)])
set(["/".join(blob.name.split('/',maxsplit=2)[0:2]) for blob in client.list_blobs(BUCKET_NAME, prefix=PREFIX)])
gcs = gcsfs.GCSFileSystem()
gcs.ls(BUCKET_NAME)
xyz文件夹内的文件夹，但是命令gsutil ls gs://abc/xyz返回xyz中的所有对象，包括非文件夹项。那么，你要求的是什么？所有文件夹还是所有项目，包括文件夹？ - Robino