Cloud Scheduler触发Python Cloud Function失败问题排查

问题背景 #

开发了一个Python Cloud Function，功能是从YouTube API获取视频浏览量并插入BigQuery表。手动运行时一切正常，但通过Cloud Scheduler按小时HTTP触发时失败，报错：

AttributeError: 'Request' object has no attribute 'playlistItems'

Cloud Scheduler配置：

• 目标类型：HTTP
• URL：Cloud Function的触发URL
• HTTP方法：POST
• 认证方式：无认证

错误原因 #

当通过HTTP触发Cloud Function时，functions_framework会自动将Request请求对象传入被识别为入口的函数。原代码中直接在全局作用域调用insert_rows(youtube)，但当Cloud Scheduler触发时，insert_rows被误作为HTTP请求处理函数，接收的参数是Request对象而非预先构建的YouTube客户端，导致调用youtube.playlistItems()时出现属性不存在的错误。

修复方案 #

调整代码结构，明确区分手动运行和HTTP触发的入口：

1. 使用@functions_framework.http装饰HTTP触发的处理函数
2. 将手动运行的逻辑放在if __name__ == '__main__':判断中，避免HTTP触发时自动执行
3. 确保HTTP处理函数内部调用原有业务逻辑时传入正确的YouTube客户端

修改后的完整代码 #

import functions_framework
from googleapiclient.discovery import build
from google.cloud import bigquery
from datetime import datetime

client = bigquery.Client()
# Youtube Credentials
api_key= 'my api key'

channel_id = 'UCONax_0CMWVL6b7tHulEQhA'   
api_service_name = "youtube"
api_version = "v3"
youtube = build(api_service_name, api_version, developerKey=api_key)

video_ids=[]

def get_video_details(youtube):
    request = youtube.playlistItems().list(
        part='contentDetails',
        playlistId='UUONax_0CMWVL6b7tHulEQhA',
        maxResults=50,
    )
    response = request.execute()

    # Appending for the first 50 entries
    for item in response['items']:
        video_ids.append(item['contentDetails']['videoId'])

    # Going for next pages
    next_page_token = response.get('nextPageToken')
    more_pages = True

    while more_pages:
        if next_page_token is None:
            more_pages = False
        else:
            request = youtube.playlistItems().list(
                part='contentDetails',
                playlistId='UUONax_0CMWVL6b7tHulEQhA',
                maxResults=50,
                pageToken=next_page_token  # Include nextPageToken in the request
            )
            response = request.execute()

            for item in response['items']:
                video_ids.append(item['contentDetails']['videoId'])

            next_page_token = response.get('nextPageToken')

    return video_ids

def get_video_views(youtube, video_ids):
    monthly_views = {}
    for i in range(0, len(video_ids), 50):
        if i < len(video_ids):
            request = youtube.videos().list(
                part="snippet,contentDetails,statistics",
                id=','.join(video_ids[i:i + 50])
            )
            response = request.execute()

            for video in response['items']:
                published_date = video['snippet']['publishedAt']
                date_obj = datetime.strptime(published_date, "%Y-%m-%dT%H:%M:%SZ")
                month_year = date_obj.strftime("%Y-%m")
                view_count = int(video['statistics'].get('viewCount', 0))

                if month_year in monthly_views:
                   monthly_views[month_year] += view_count
                else:
                   monthly_views[month_year] = view_count
    monthly_views_list = []
    for month, views in monthly_views.items():
       monthly_views_list.append({'month': month, 'views': views})

    return monthly_views_list

def insert_rows(youtube):
    video_ids = get_video_details(youtube)
    monthly_views = get_video_views(youtube, video_ids)

    table_id = "trans-market-427605-f1.Dummy_Test.Dummy_Test_Python1"
    errors = client.insert_rows_json(table_id, monthly_views)
    if errors == []:
       print("New rows have been added.")
    else:
       print("Encountered errors while inserting rows: {}".format(errors))

# HTTP触发入口
@functions_framework.http
def handle_request(request):
    # 忽略请求内容，直接执行业务逻辑
    insert_rows(youtube)
    return "Success", 200

# 手动运行入口
if __name__ == '__main__':
    insert_rows(youtube)

关键修改说明 #

• 添加@functions_framework.http装饰的handle_request函数，作为HTTP触发的入口，该函数接收Request对象，内部调用insert_rows(youtube)传入正确的客户端
• 将原全局作用域的insert_rows(youtube)调用移到if __name__ == '__main__':块中，仅在手动运行时执行
• 确保HTTP触发时，业务逻辑使用的是预先构建的YouTube客户端，而非Request对象

验证 #

部署修改后的Cloud Function后，重新通过Cloud Scheduler触发，即可正常执行数据插入逻辑。

内容的提问来源于stack exchange，提问作者Praveen Kumar