repository·main·Indexed 42 days ago
https://github.com/inngest/inngest-pyA Python SDK for building durable, event-driven background jobs and workflows without managing queuing infrastructure. It supports Python 3.10+ and integrates with frameworks including FastAPI (>=0.110.0), Django (>=5.0), Flask (>=3.0.0), Tornado (>=6.4), and DigitalOcean Functions. Features include durable steps via step.run, automatic retries, state management, and an EncryptionMiddleware for securing event data and function inputs/outputs.
The Inngest Python SDK allows you to add durable functions, background jobs, and scheduled jobs to your Python applications without managing separate queuing infrastructure like Celery. It enables writing background jobs as step functions that are resilient to failures.
Supported Frameworks:
>=5.0)>=0.110.0)>=3.0.0)>=6.4)Requirements:
When the Inngest server triggers a function, it sends a POST request to the SDK (e.g., POST /api/inngest?fnId=...&stepId=...).
The Execution Process:
fnId (the function identifier) and stepId (the specific step to target, if applicable) from the query parameters.ServerRequest from the request body. This object contains the triggering event, event batch, memoized step outputs, and execution context (such as run_id and attempt).use_api=true. In this case, the SDK fetches the event batch and step memos from the Inngest API instead of reading them from the request body.Before the Inngest server can execute functions, it must know their configurations (triggers, concurrency limits, retry policies, etc.). This process is called Syncing.
Syncing is triggered when the Inngest server sends a PUT request to the SDK's serve endpoint. The mechanism used is determined by the x-inngest-sync-kind header. The SDK supports two modes:
PUT request with function configs and app metadata. This is the preferred mechanism and requires request signature verification.POST to the Inngest API's /fn/register endpoint using the signing key hash as a Bearer token, then forwards that response back to the original PUT caller.The Connect mechanism uses a WebSocket-based handshake instead of standard HTTP PUT requests.
During the handshake, the SDK sends a WORKER_CONNECT protobuf message containing the JSON-encoded function configurations. When using Connect, the runtime URLs use the wss://connect scheme instead of http, and the runtime type is identified as ws rather than http.
The Connect implementation uses a dual-thread model to ensure that communication with the Inngest server is never blocked by user code:
asyncio.run_coroutine_threadsafe().CommHandler and handles OS signals (like SIGTERM) for graceful shutdown.If a WebSocket connection drops after a worker completes an execution but before the reply is acknowledged, the SDK uses Flushing to ensure the result is delivered.
Unacknowledged replies are sent via HTTP to the /v0/connect/flush endpoint. A periodic poller automatically checks for these unacked replies and flushes them. To prevent Out-Of-Memory (OOM) risks, unacked replies are held in a buffer rather than being kept in memory indefinitely.
The Connect implementation follows a specific lifecycle to maintain a live connection between the SDK and the Inngest server:
/v0/connect/start via REST to obtain a WebSocket URL, connection ID, and auth tokens.WORKER_CONNECT message containing function configurations and app metadata, then waits for a ready signal from the server.SIGTERM or SIGINT, the SDK sends a pause message, waits for in-flight requests to drain, and then closes.The Inngest Python SDK uses signing keys to authenticate requests between the Inngest server and your application. It supports two types of keys to facilitate zero-downtime rotation:
Note on Environments:
{parent_id}-failure. These handlers are triggered by the internal inngest/function.failed event, filtered by the parent function's ID.Parallel execution allows multiple steps to run concurrently on the Inngest Executor. This is achieved through a discovery phase using Group.parallel():
in_parallel context variable to True.PLANNED and a ResponseInterrupt is raised.ResponseInterrupt.stepId query parameter to tell the SDK which specific step to execute during subsequent replays, while other steps in the group raise a SkipInterrupt to be ignored.Inngest functions use a deterministic replay model. An Inngest function is called multiple times per run—once for each step (or group of parallel steps) that needs to execute.
Each time the function is called, it replays from the top. The SDK uses StepMemos (memoized results) to skip steps that have already completed. When the SDK encounters a step with no memo, it either executes the step or 'plans' it (for parallel or deferred execution) and then interrupts the function to return control to the Inngest server.
**Key behaviors:
@client.create_function(
fn_id="my-fn",
trigger=inngest.TriggerEvent(event="app/signup"),
)
async def my_fn(ctx: inngest.Context) -> str:
user = await ctx.step.run("create-user", create_user) # Step A
await ctx.step.run("send-email", lambda: send_email(user)) # Step B
return "done"The Inngest server (the "Executor") drives function execution by sending HTTP requests or WebSocket messages to the SDK. The SDK is reactive and only responds to these requests.
The Replay Model A single function run typically involves multiple requests. Each request replays the function from the top. To ensure efficiency and durability, the SDK uses memoized step results from prior requests to skip work that has already been completed. This replay-based model is the core of Inngest's durable execution.
Execution Flow
flask.py, fast_api.py, django.py). These adapters normalize the framework's request into a CommRequest.CommHandler, which acts as the central dispatcher.POST requests, the SDK deserializes a ServerRequest containing the event, memoized step outputs, and execution context (run ID, attempt number).CommResponse and sends it back to the Executor.