Pydantic

This guide will help you use Pydantic to perform runtime type validation when sending and receiving events.

Step output

Steps can return Pydantic objects as long as the output_type parameter is set to the Pydantic model return type.

client = inngest.Inngest(
    app_id="my-app",

    # Must set the client serializer when using Pydantic output
    serializer=inngest.PydanticSerializer(),
)

class User(pydantic.BaseModel):
    name: str

async def get_user() -> User:
    return User(name="Alice")

@client.create_function(
    fn_id="my-fn",
    trigger=inngest.TriggerEvent(event="my-event"),
)
async def my_fn(ctx: inngest.Context) -> None:
    # user object is a Pydantic object at both runtime and compile time
    user = await ctx.step.run("get-user", get_user, output_type=User)

More complex types work as well. For example, if the get_user function returned an list[Admin | User] type, you could set the output_type to list[Admin | User].

await ctx.step.run("get-person", get_users, output_type=list[User | Admin])

Why do I need to set the output_type parameter?

Since step output is transmitted as JSON back to the Inngest server, we lose the reference to the original Python class. So the output_type parameter is used to deserialize the JSON back into the correct type.

Why can't the SDK infer the output type from my type annotations?

The could sometimes work, but there are common patterns that break it. Runtime return type inference is impossible if the return type:

  • Is implicit (i.e. not specified but your type checker figures it out).
  • Is a generic.

Function output

Functions can return Pydantic objects as long as the output_type parameter is set to the Pydantic model return type.

client = inngest.Inngest(
    app_id="my-app",

    # Must set the client serializer when using Pydantic output
    serializer=inngest.PydanticSerializer(),
)

class User(pydantic.BaseModel):
    name: str

@client.create_function(
    fn_id="my-fn",
    output_type=User,
    trigger=inngest.TriggerEvent(event="my-event"),
)
async def my_fn(ctx: inngest.Context) -> None:
    return User(name="Alice")

Sending events

Create a base class that all your event classes will inherit from. This class has methods to convert to and from inngest.Event objects.

import inngest
import pydantic
import typing

TEvent = typing.TypeVar("TEvent", bound="BaseEvent")

class BaseEvent(pydantic.BaseModel):
    data: pydantic.BaseModel
    id: str = ""
    name: typing.ClassVar[str]
    ts: int = 0

    @classmethod
    def from_event(cls: type[TEvent], event: inngest.Event) -> TEvent:
        return cls.model_validate(event.model_dump(mode="json"))

    def to_event(self) -> inngest.Event:
        return inngest.Event(
            name=self.name,
            data=self.data.model_dump(mode="json"),
            id=self.id,
            ts=self.ts,
        )

Next, create a Pydantic model for your event.

class PostUpvotedEventData(pydantic.BaseModel):
    count: int

class PostUpvotedEvent(BaseEvent):
    data: PostUpvotedEventData
    name: typing.ClassVar[str] = "forum/post.upvoted"

Since Pydantic validates on instantiation, the following code will raise an error if the data is invalid.

client.send(
    PostUpvotedEvent(
        data=PostUpvotedEventData(count="bad data"),
    ).to_event()
)

Receiving events

When defining your Inngest function, use the name class field when specifying the trigger. Within the function body, call the from_event class method to convert the inngest.Event object to your Pydantic model.

@client.create_function(
    fn_id="handle-upvoted-post",
    trigger=inngest.TriggerEvent(event=PostUpvotedEvent.name),
)
def fn(ctx: inngest.ContextSync) -> None:
    event = PostUpvotedEvent.from_event(ctx.event)