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)