Cloud Functions でPub/Subから引数を受け取って発火する関数を
ローカルで作成したのでメモ

環境

Ubuntu 22.04LTS
python 3.10.4

準備

$ pip install functions-framework

流れ

main.pyを作成
ターミナル（A）でfunctions-frameworkを起動する
別のターミナル（B）を立ち上げ、curlコマンドでイベントを発行
ターミナル（A）に、main.pyで処理された結果が表示される

詳細

1. main.pyを作成する

.
└── main.py

main.py

# [START functions_helloworld_pubsub]
# 関数 hello_pubsub に、event と context という2つの引数がPub/Subから渡される。
# 構造はコメントに書いてある構成になっている。
def hello_pubsub(event, context):
    """Background Cloud Function to be triggered by Pub/Sub.
    Args:
         event (dict):  The dictionary with data specific to this type of　event. 
　　　　　　　           The `@type` field maps to　`type.googleapis.com/google.pubsub.v1.PubsubMessage`.
                        The `data` field maps to the PubsubMessage data in a base64-encoded string. The `attributes` field maps
                        to the PubsubMessage attributes if any is present.
         context (google.cloud.functions.Context): Metadata of triggering event                    including `event_id` which maps to the PubsubMessage
                        messageId, `timestamp` which maps to the PubsubMessage
                        publishTime, `event_type` which maps to
                        `google.pubsub.topic.publish`, and `resource` which is
                        a dictionary that describes the service API endpoint
                        pubsub.googleapis.com, the triggering topic's name, and
                        the triggering event type
                        `type.googleapis.com/google.pubsub.v1.PubsubMessage`.
    Returns:
        None. The output is written to Cloud Logging.
    """
    import base64

    print("""This Function was triggered by messageId {} published at {} to {}
    """.format(context.event_id, context.timestamp, context.resource["name"]))

    if 'data' in event:
        name = base64.b64decode(event['data']).decode('utf-8')
    else:
        name = 'World'
    print('Hello {}!'.format(name))
# [END functions_helloworld_pubsub]

2. ターミナル（A）で functions-framework を起動する

実行後に止まったように見えますが、正常です

$ functions-framework --target=hello_pubsub --port=8080 --signature-type=event

3. 別のターミナル（B）を起動し、curlコマンドを投げる。

# 下から3行目の "data": "d29ybGQ=" は、単語 'world'をbase64でエンコードしたもの
$ curl localhost:8080 \
  -X POST \
  -H "Content-Type: application/json" \
  -d '{
        "context": {
          "eventId":"1144231683168617",
          "timestamp":"2020-05-06T07:33:34.556Z",
          "eventType":"google.pubsub.topic.publish",
          "resource":{
            "service":"pubsub.googleapis.com",
            "name":"projects/sample-project/topics/gcf-test",
            "type":"type.googleapis.com/google.pubsub.v1.PubsubMessage"
          }
        },
        "data": {
          "@type": "type.googleapis.com/google.pubsub.v1.PubsubMessage",
          "attributes": {
             "attr1":"attr1-value"
          },
          "data": "d29ybGQ="
        }
      }'

4. ターミナル（A）に結果が表示される

This Function was triggered by messageId 1144231683168617 published at 2020-05-06T07:33:34.556Z to projects/sample-project/topics/gcf-test
    
Hello This is the TEST!

5. コンソールでテストする場合

コンソールで動かす場合は、関数をデプロイしたのちに「テスト中」タブを選択。
{}部分に上記のJSONを記入、「関数をテストする」を押下すると、画面下部に結果が出力される。

参考

Cloud Functions ローカルでの開発-ローカル関数の呼び出し
 Cloud Functions Cloud Pub/Sub トリガー