def marshal_background_event_data()

in src/functions_framework/event_conversion.py [0:0]

• 26 lines of code
• 3 McCabe index (conditional complexity)

def marshal_background_event_data(request):
    """Marshal the request body of a raw Pub/Sub HTTP request into the schema that is expected of
    a background event"""
    try:
        request_data = request.get_json()
        if not _is_raw_pubsub_payload(request_data):
            # If this in not a raw Pub/Sub request, return the unaltered request data.
            return request_data
        return {
            "context": {
                "eventId": request_data["message"]["messageId"],
                "timestamp": request_data["message"].get(
                    "publishTime", datetime.utcnow().isoformat() + "Z"
                ),
                "eventType": _PUBSUB_EVENT_TYPE,
                "resource": {
                    "service": _PUBSUB_CE_SERVICE,
                    "type": _PUBSUB_MESSAGE_TYPE,
                    "name": _parse_pubsub_topic(request.path),
                },
            },
            "data": {
                "@type": _PUBSUB_MESSAGE_TYPE,
                "data": request_data["message"]["data"],
                "attributes": request_data["message"].get("attributes", {}),
            },
        }
    except (AttributeError, KeyError, TypeError):
        raise EventConversionException("Failed to convert Pub/Sub payload to event")