ReferencesPython SDKSteps

Invoke by ID v0.3.0+

Calls another Inngest function, waits for its completion, and returns its output.

This method behaves identically to the invoke step method, but accepts an ID instead of the function object. This can be useful for a few reasons:

  • Trigger a function whose code is in a different codebase.
  • Avoid circular dependencies.
  • Avoid undesired transitive imports.

Arguments

  • Name
    step_id
    Type
    str
    Required
    required
    Description

    Step ID. Should be unique within the function.

  • Name
    app_id
    Type
    str
    Required
    optional
    Description

    App ID of the invoked function.

  • Name
    function_id
    Type
    str
    Required
    required
    Description

    ID of the invoked function.

  • Name
    data
    Type
    object
    Required
    optional
    Description

    JSON-serializable data that will be passed to the invoked function as event.data.

  • Name
    user
    Type
    object
    Required
    optional
    Description

    JSON-serializable data that will be passed to the invoked function as event.user.

Examples

Within the same app

@inngest_client.create_function(
    fn_id="fn-1",
    trigger=inngest.TriggerEvent(event="app/fn-1"),
)
async def fn_1(
    ctx: inngest.Context,
    step: inngest.Step,
) -> str:
    return "Hello!"

@inngest_client.create_function(
    fn_id="fn-2",
    trigger=inngest.TriggerEvent(event="app/fn-2"),
)
async def fn_2(
    ctx: inngest.Context,
    step: inngest.Step,
) -> None:
    output = step.invoke_by_id(
        "invoke",
        function_id="fn-1",
    )

    # Prints "Hello!"
    print(output)

Across apps

inngest_client_1 = inngest.Inngest(app_id="app-1")
inngest_client_2 = inngest.Inngest(app_id="app-2")

@inngest_client_1.create_function(
    fn_id="fn-1",
    trigger=inngest.TriggerEvent(event="app/fn-1"),
)
async def fn_1(
    ctx: inngest.Context,
    step: inngest.Step,
) -> str:
    return "Hello!"

@inngest_client_2.create_function(
    fn_id="fn-2",
    trigger=inngest.TriggerEvent(event="app/fn-2"),
)
async def fn_2(
    ctx: inngest.Context,
    step: inngest.Step,
) -> None:
    output = step.invoke_by_id(
        "invoke",
        app_id="app-1",
        function_id="fn-1",
    )

    # Prints "Hello!"
    print(output)