如何从事件触发的云功能中检索 delivery_attempt？

Question

我正在编写一个 Python 云函数，我想检索“delivery_attempt”属性。

根据文档，Cloud Function 仅获取 2 个参数：事件（PubsubMessage 类型）和上下文。

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.
    """

如何检索邮件的delivery_attempt？

Answer 1

通过 Pub/Sub 触发云函数时，无法访问传递尝试字段。相反，获取访问权限的方法是设置一个基于 HTTP 的云函数，并在您单独创建的订阅中使用触发 URL 作为推送端点。

然后，您可以按如下方式访问传递尝试：

def hello_world(request):
    """Responds to any HTTP request.
    Args:
        request (flask.Request): HTTP request object.
    Returns:
        The response text or any set of values that can be turned into a
        Response object using
        `make_response <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
    """
    request_json = request.get_json()
    print(request_json["deliveryAttempt"])