Using microsoft azure eventgrid trigger output binding, how to support cloudEvent Schema 01

Question

I'm using Azure event grid trigger output binding python v2 model, the trigger works fine, the events type is cloudSchema. For the output i keep getting

message": "This resource is configured to receive event in 'CloudEventV10' schema. The JSON received does not conform to the expected schema. Token Expected: StartObject, Actual Token Received: StartArray

import logging
import azure.functions as func
import datetime

@app.function_name(name="eventgrid_output")
@app.route(route="eventgrid_output")
@app.event_grid_output(
    arg_name="outputEvent",
    topic_endpoint_uri="MyEventGridTopicUriSetting",
    topic_key_setting="MyEventGridTopicKeySetting")
def eventgrid_output(eventGridEvent: func.EventGridEvent, 
         outputEvent: func.Out[func.EventGridOutputEvent]) -> None:

    logging.log("eventGridEvent: ", eventGridEvent)

    outputEvent.set(
        func.EventGridOutputEvent(
            id="test-id",
            data={"tag1": "value1", "tag2": "value2"},
            subject="test-subject",
            event_type="test-event-1",
            event_time=datetime.datetime.utcnow(),
            data_version="1.0"))

How do i set the output type to CloudEventV10

Successfully publishing the event triggered the function

Answer 1

This worked for me: You are not using Event hub Trigger but you are using its arg name in the function which is eventGridEvent: func.EventGridEvent,, which is resulting in error

Exception: FunctionLoadError: cannot load the test function: the following parameters are declared in Python but not in function.json: {'eventGridEvent'}

I have used EventHub Trigger and Event Hub output binding in same trigger to get output.

For reference check this document

My Code:

import azure.functions as func
import datetime
import logging

app = func.FunctionApp()

@app.function_name(name="eventgrid_out")
@app.event_grid_trigger(arg_name="eventGridEvent")
@app.event_grid_output(
    arg_name="outputEvent",
    topic_endpoint_uri="MyEventGridTopicUriSetting",
    topic_key_setting="MyEventGridTopicKeySetting")
def eventgrid_output(eventGridEvent: func.EventGridEvent, 
         outputEvent: func.Out[func.EventGridOutputEvent]) -> None:

    logging.info("eventGridEvent: %s", eventGridEvent)

    outputEvent.set(
        func.EventGridOutputEvent(
            id="test-id",
            data={"tag1": "value1", "tag2": "value2"},
            subject="test-subject",
            event_type="test-event-1",
            event_time=datetime.datetime.utcnow(),
            data_version="1.0"))

OUTPUT:

To check the output binding data sent to event grid, I used a service bus queue.

I think it is not mention properly in document, I have created Document correction issue. follow the issue for further updates.