列出不属于 Cloud Storage 中文件夹的 Blob

Question

我正在尝试使用official Python client library 列出所有未存储在特定文件夹中的 blob（在我的情况下为文件），例如 not-target。

我知道库函数提供了prefix 选项，以便按所需路径过滤 blob。但是，我想在这里做相反的事情 - 只列出那些在提供的路径中不存在的 blob。

当前解决方案

• 使用 list_blobs 遍历所有 blob
• 按名称过滤掉以列入黑名单的 not-target 路径开头的 blob。

for blob in bucket.list_blobs():
    if not blob.name.startswith('not-target/'):
        # process the blob    

不足之处

• 需要很长时间，因为存储桶中的大多数 blob 都属于此文件夹。

是否有我不知道的针对此任务的优化？

Answer 1

这种方法可以通过 list_blobs 方法的 prefix 和 delimiter 参数来完成。

例如，如果你想列出所有在root中但不在一个文件夹中的对象，你可以使用下面的代码sn-p：

from google.cloud import storage

client = storage.Client()
bucket = client.get_bucket("my_bucket")
all_blobs = list(client.list_blobs(bucket, prefix="", delimiter="/"))

for blob in all_blobs:
    print (blob.name)

另外，请注意，在 root 中列出时，不需要添加 prefix 参数：

from google.cloud import storage

client = storage.Client()
bucket = client.get_bucket("my_bucket")
all_blobs = list(client.list_blobs(bucket, delimiter="/"))

for blob in all_blobs:
    print (blob.name)

例如，另一种方法可能是列出文件夹内的文件，但不递归地列出，即该文件夹根目录中的对象。那么代码可能是：

from google.cloud import storage

client = storage.Client()
bucket = client.get_bucket("my_bucket")
all_blobs = list(client.list_blobs(bucket,prefix="folder/", delimiter="/"))

for blob in all_blobs:
    print (blob.name)

您可以根据列出时使用的路径更改prefix。

如果您可能想要列出除特定文件夹之外的所有内容，则没有直接的方法可以使用 Cloud Storage 库来实现。在这种情况下，我认为最好的方法是在客户端过滤数据。您可以使用正则表达式过滤您在Storage中列出后得到的列表：

from google.cloud import storage
import re
regex = re.compile(r'folder.*')

client = storage.Client()
bucket = client.get_bucket("my_bucket")
all_blobs = list(client.list_blobs(bucket))

filtered = [blob for blob in all_blobs if not regex.match(blob.name)]

for blob in filtered:
    print(blob.name)

Answer 2

为了重申 OP 的问题，我们有一个顶级文件夹 "not-target/"，其中包含许多文件，这意味着遍历它们的效率非常低。但是，我们希望列出所有其他 blob，包括顶级（也称为根文件夹）和任何不以 "not-target/" 开头的子文件夹（递归）中的 blob。

为此，我们将做三件事：

1. 在根（顶级）级别获取 blob - 它们不在任何文件夹中，因此不在"not-target" 中。
2. 获取根级别的文件夹，忽略文件夹"not-target"。
3. 对于每个文件夹，获取所有（递归）包含的 blob。
4. 打印出所有 blob 名称（即路径）。

代码：

import google.cloud.storage as gcs
BUCKET_NAME = "my-cloud-bucket" # The name for bucket gs://my-cloud-bucket
skip_folders = {"not-target/"} # Folders need to have trailing slash "/"
client = gcs.Client()
root_iter = client.list_blobs(
    bucket_or_name=BUCKET_NAME,
    delimiter="/",
)
root_blobs = list(root_iter)
root_folders = root_iter.prefixes
sub_blobs = []
for folder in root_folders:
    if folder in skip_folders:
        continue
    sub_blobs += list(client.list_blobs(
        bucket_or_name=BUCKET_NAME,
        prefix=folder
    ))
for blob in root_blobs + sub_blobs:
    print(blob.name)

需要考虑的一些注意事项：

1. 由于我们在根级别工作，我们不需要指定 kwarg prefix=""，但您可以。
2. 对于root_iter，我们仅需要此级别的 blob，因此请指定 kwarg delimiter="/"。
3. 在访问root_iter.prefixes 之前，您需要遍历root_iter。此对象是延迟加载的，这意味着在对象进行 api 调用之前不会填充此成员，为此您需要至少获取第一个元素。我们想要所有元素（根级别的 blob），所以 list(root_iter) 为我们做这件事。
4. root_iter.prefixes 为我们获取根级别的文件夹 - 带有尾部斜杠 ".../"。
5. 对于 sub_blobs，我们希望所有子文件夹中的所有 blob 递归，因此在这种情况下，我们不提供 delimiter kwarg。