在GCP中使用Python Pub/Sub客户端如何过滤订阅列表子集？

gcloud CLI的过滤方式 #

在gcloud CLI中，列出项目内Pub/Sub订阅时可直接通过--filter参数在客户端层面过滤结果，示例命令：

gcloud --project=my-project pubsub subscriptions list --filter=my-filter-string --format='value(name)'

Python客户端当前的本地过滤方案 #

使用Google Cloud Pub/Sub Python客户端的list_subscriptions方法时，该方法仅支持指定项目参数，会返回项目内所有订阅，因此只能先全量获取再本地过滤，示例代码如下：

from google.cloud import pubsub_v1

subscriber_client = pubsub_v1.SubscriberClient()

filter_str = "my-filter-string"  # 避免与Python内置filter关键字冲突

with subscriber_client:
    page_result = subscriber_client.list_subscriptions(
            project="projects/my-project",
        )
    filtered_subscriptions = [
        subscription.name
        for subscription in page_result
        if filter_str in subscription.name.split("/")[-1]
    ]
    for subscription_name in filtered_subscriptions:
        print(subscription_name)

更高效的替代方案 #

核心说明 #

Pub/Sub的ListSubscriptions原生API本身不支持服务端过滤，gcloud CLI的--filter本质也是客户端层面的过滤逻辑（和上述本地过滤一致）。Python客户端的metadata参数用于传递请求头元数据（如认证信息），与过滤功能无关，无法通过它实现服务端筛选。

优化方向 #

1. 基于Cloud Asset Inventory API实现服务端过滤
   如果订阅数量庞大，全量获取后过滤效率低下，可以借助Cloud Asset Inventory API，通过标签或属性筛选订阅，实现服务端级别的查询，示例代码：

from google.cloud import asset_v1

client = asset_v1.AssetServiceClient()
project_scope = "projects/my-project"
# 按标签过滤，示例为筛选标签env=prod的订阅
query_filter = 'labels.env="prod" AND assetType="pubsub.googleapis.com/Subscription"'

response = client.search_all_resources(
    request={"scope": project_scope, "query": query_filter}
)

for resource in response:
    print(resource.name)

注意：使用该API需要具备cloudasset.assets.searchAllResources权限，且查询结果基于资产快照，可能存在一定延迟。

2. 优化本地过滤逻辑
   如果仅按名称包含特定字符串过滤，当前的分页遍历过滤逻辑已经足够高效——page_result是分页迭代器，不会一次性加载所有订阅到内存，内存占用可控，无需额外优化。

内容的提问来源于stack exchange，提问作者Emmanuel Sciara