如何从Google Cloud Logging API获取next_page_token？解决生成器对象错误

错误原因 #

loggingClient.list_entries()默认返回的是生成器对象，它会自动遍历所有分页条目，但生成器本身没有next_page_token属性，直接访问会抛出AttributeError。

解决方案 #

方案1：自动遍历所有日志（无需手动处理token） #

如果只是需要获取当月所有日志，直接迭代生成器即可，它会自动处理分页逻辑：

from pytz import timezone
from datetime import datetime
from calendar import monthrange
from google.cloud import logging
from google.oauth2 import service_account

def getInstanceLogsList(
    instanceId: str,
    authTokenDict: dict,
):
    today = datetime.now().astimezone(tz=timezone("Asia/Kolkata"))

    monthStartDay = 1
    monthEndDay = monthrange(today.year, today.month)[1]
    
    credentials = service_account.Credentials.from_service_account_info(authTokenDict)
    loggingClient = logging.Client(credentials=credentials)

    # 生成符合RFC3339标准的时间字符串（GCP Logging要求的格式）
    end_dt = datetime(year=today.year, month=today.month, day=monthEndDay, hour=23, minute=59, second=59, tzinfo=timezone("Asia/Kolkata"))
    start_dt = datetime(year=today.year, month=today.month, day=monthStartDay, hour=0, minute=0, second=0, tzinfo=timezone("Asia/Kolkata"))
    startDateTimeObject = start_dt.isoformat()
    endDateTimeObject = end_dt.isoformat()

    listLogsResponse = []
    # 直接迭代生成器，自动处理分页
    for entry in loggingClient.list_entries(
        filter_=f'resource.type = "gce_instance" AND resource.labels.instance_id = "{instanceId}" AND timestamp >= "{startDateTimeObject}" AND timestamp <= "{endDateTimeObject}"'
    ):
        listLogsResponse.append(entry)
    
    return listLogsResponse

方案2：手动获取next_page_token（需保存token时使用） #

如果需要手动控制分页并获取next_page_token，需要通过pages属性访问分页响应对象：

from pytz import timezone
from datetime import datetime
from calendar import monthrange
from google.cloud import logging
from google.oauth2 import service_account

def getInstanceLogsList(
    instanceId: str,
    authTokenDict: dict,
):
    today = datetime.now().astimezone(tz=timezone("Asia/Kolkata"))

    monthStartDay = 1
    monthEndDay = monthrange(today.year, today.month)[1]
    
    credentials = service_account.Credentials.from_service_account_info(authTokenDict)
    loggingClient = logging.Client(credentials=credentials)

    # 生成符合RFC3339标准的时间字符串
    end_dt = datetime(year=today.year, month=today.month, day=monthEndDay, hour=23, minute=59, second=59, tzinfo=timezone("Asia/Kolkata"))
    start_dt = datetime(year=today.year, month=today.month, day=monthStartDay, hour=0, minute=0, second=0, tzinfo=timezone("Asia/Kolkata"))
    startDateTimeObject = start_dt.isoformat()
    endDateTimeObject = end_dt.isoformat()

    pageToken = None
    listLogsResponse = []
    
    while True:
        # 获取分页迭代器
        iterator = loggingClient.list_entries(
            filter_=f'resource.type = "gce_instance" AND resource.labels.instance_id = "{instanceId}" AND timestamp >= "{startDateTimeObject}" AND timestamp <= "{endDateTimeObject}"',
            page_token=pageToken
        ).pages
        
        try:
            page = next(iterator)
            listLogsResponse.extend(page)
            pageToken = page.next_page_token
            if not pageToken:
                break
        except StopIteration:
            break
    
    return listLogsResponse

额外注意事项 #

原代码中用strftime("%Y-%m-%dT%H:%M:%S%z")生成的时区格式是+0530，而GCP Logging的timestamp筛选要求RFC3339格式（如+05:30），改用isoformat()可以避免筛选条件失效。

内容的提问来源于stack exchange，提问作者Daniel Vanum