Cloud Run Functionsを使ってBigQueryを操作したいのですが、エラーが出て解決しません。

Question

実現したいこと

Cloud Run Functionsを使って、Cloud StorageへCSVはアップロードをトリガーとして、CSV内容をBigQueryにインポートしたい。

発生している問題・分からないこと

引数contextにデータが入らないとなってしまいます。

エラーメッセージ

TypeError: process_csv_to_bigquery() missing 1 required positional argument: 'context'

該当のソースコード

import pandas as pd
from google.cloud import bigquery, storage
from io import StringIO
import os
import datetime

def process_csv_to_bigquery(event, context):
    """
    Google Cloud Storageにファイルがアップロードされた際にトリガーされる関数。
    CSVを読み込み、データを整形し、BigQueryに挿入します。
    """

    file_name = event['name']
    bucket_name = event['bucket']

    print(f"Processing file: {file_name} from bucket: {bucket_name}")

    # CSVファイル以外はスキップ
    if not file_name.endswith('.csv'):
        print(f"Skipping non-CSV file: {file_name}")
        return

    # GCSからCSVファイルを読み込む
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(file_name)
    csv_data = blob.download_as_string().decode('utf-8')

    # pandasでCSVデータをDataFrameとして読み込む
    df = pd.read_csv(StringIO(csv_data))

    # 日付形式の整形 (年/月/日 -> 年-月-日) とデータ型をDATEにする
    if '日付' in df.columns:
        # まずPandasのdatetime型に変換。変換できない値はNaT（Not a Time）にする
        df['日付'] = pd.to_datetime(df['日付'], format='%Y/%m/%d', errors='coerce')
        # NaTではない有効な日付のみをdatetime.dateオブジェクトに変換し、文字列化
        # BigQueryのDATE型として挿入するために 'YYYY-MM-DD' 形式の文字列にします。
        df['日付'] = df['日付'].dt.date.astype(str)
        # NaTだった場合はNaNになるので、BigQueryに挿入される際にNULLとして扱われます。

    # 時間形式の整形 (hh:mm -> hh:mm:ss) とデータ型をTIMEにする
    time_columns = ['支払開始時間', '支払終了時間', '請求開始時間', '請求終了時間']
    for col in time_columns:
        if col in df.columns:
            # NaN値ではない、かつ形式が 'HH:MM'（長さ5）の場合に ':00' を追加
            df[col] = df[col].astype(str).apply(lambda x: x + ':00' if pd.notna(x) and len(x) == 5 else x)

            # BigQueryのTIME型として挿入するため、datetime.timeオブジェクトに変換を試みます。
            # 変換できない場合はNaTになり、最終的にNULLとして扱われます。
            df[col] = df[col].apply(lambda x: datetime.datetime.strptime(x, '%H:%M:%S').time() if pd.notna(x) and len(x) == 8 else None)
            # BigQueryのTIME型は 'HH:MM:SS' 形式の文字列も受け入れることが多いですが、
            # datetime.time オブジェクトに変換することでより厳密に型を扱えます。

    # BigQueryクライアントの初期化
    bigquery_client = bigquery.Client()

    # BigQueryプロジェクトID、データセットID、テーブルIDを設定
    # ここをあなたのBigQueryのデータセットIDとテーブルIDに置き換えてください。
    project_id = bigquery_client.project # BigQueryクライアントが自動的にプロジェクトIDを検出
    dataset_id = 'canvas-eye-465212-q8.zinzai_test_date'  # 例: 'my_dataset' **ここをあなたのデータセットIDに置き換えてください**
    table_id = 'canvas-eye-465212-q8.zinzai_test_date.worklog'      # 例: 'my_table'    **ここをあなたのテーブルIDに置き換えてください**

    table_ref = bigquery_client.dataset(dataset_id, project=project_id).table(table_id)

    # BigQueryから既存データの削除 (CSVに含まれる日付の最小値と最大値に該当するデータ)
    # CSVに「日付」カラムがあり、かつデータが存在する場合のみ削除処理を実行
    if '日付' in df.columns and not df['日付'].empty:
        # 有効な日付データ（NaNではないデータ）のみを考慮
        valid_dates_in_csv = df['日付'].dropna()
        if not valid_dates_in_csv.empty:
            min_date_str = valid_dates_in_csv.min()
            max_date_str = valid_dates_in_csv.max()

            # BigQueryのDATE型カラムに対して日付範囲でDELETEを実行
            delete_query = f"""
            DELETE FROM `{project_id}.{dataset_id}.{table_id}`
            WHERE 日付 BETWEEN PARSE_DATE('%Y-%m-%d', '{min_date_str}') AND PARSE_DATE('%Y-%m-%d', '{max_date_str}')
            """
            print(f"Executing delete query: {delete_query}")
            try:
                query_job = bigquery_client.query(delete_query)
                query_job.result() # クエリの完了を待機
                print(f"Successfully deleted existing data in BigQuery for dates between {min_date_str} and {max_date_str}.")
            except Exception as e:
                print(f"Error during BigQuery delete operation: {e}")
                # 削除に失敗しても、データ挿入は試みます。必要に応じてここで処理を中断することも可能です。
        else:
            print("No valid '日付' found in the CSV for deletion range. Skipping deletion.")
    else:
        print("CSV does not contain '日付' column or it is empty. Skipping deletion.")

    # BigQueryへのデータ挿入
    # DataFrameを辞書のリストに変換します。BigQueryのスキーマとカラム数・並び順が完全に一致しているため、
    # BigQueryの insert_rows_json メソッドが最もシンプルで効果的です。
    rows_to_insert = df.to_dict(orient='records')

    print(f"Attempting to insert {len(rows_to_insert)} rows into BigQuery table: {table_id}")
    errors = bigquery_client.insert_rows_json(table_ref, rows_to_insert)

    if errors:
        print("Encountered errors while inserting rows:")
        for error in errors:
            print(f"  {error}") # エラーの詳細を表示
        raise Exception("BigQuery insert failed. Check logs for details.") # 挿入に失敗したら関数を終了させる
    else:
        print(f"Successfully inserted {len(df)} rows into BigQuery table: {table_id}")

    print("CSV processing and BigQuery insertion complete.")

試したこと・調べたこと

• teratailやGoogle等で検索した
• ソースコードを自分なりに変更した
• 知人に聞いた
• その他

上記の詳細・結果

Pythonのバージョン変更などをしましたが、なぜ間違っているか全くわからず・・・。

補足

色々調べた結果にたどり着いたコードですが、考え方自体が間違っている場合は、教えて頂きたいです。

Comments

[68user]

これって第一世代 (gen1) の書き方なような気がしますが、今作るなら第二世代 (gen2) でよいでしょう。
最初は下記を見ながら Hello world から始めるのがよいのでは。
https://cloud.google.com/run/docs/write-http-functions?hl=ja

動かない場合はソースに加えてデプロイ方法も提示してください。

[itcompass]

コメントありがとうございます。確かに、第1世代になっていました！ありがとうございます！