1

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

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

当前解决方案

  • 使用迭代所有 bloblist_blobs
  • not-target按名称过滤掉以黑名单路径开头的 blob 。
for blob in bucket.list_blobs():
    if not blob.name.startswith('not-target/'):
        # process the blob

缺点

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

我不知道这项任务是否有优化？

4

2 回答 2

3

这种方法可以使用方法的prefixanddelimiter参数来完成list_blobs。

例如，如果要列出文件夹中root但不在文件夹中的所有对象，可以使用以下代码段：

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)
于 2020-07-20T21:42:00.720 回答
0

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

为此，我们将做三件事：

  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在访问root_iter.prefixes. 此对象是延迟加载的，这意味着在对象进行 api 调用之前不会填充此成员，为此您需要至少获取第一个元素。我们想要所有元素（根级别的 blob），list(root_iter)这对我们来说也是如此。
  4. root_iter.prefixes为我们获取根级别的文件夹 - 带有斜杠".../"。
  5. 对于sub_blobs，我们希望递归所有子文件夹中的所有 blob，因此在这种情况下，我们不提供delimiterkwarg。
于 2021-10-19T10:09:13.607 回答