django-inngest 0.0.2

Missing Django integration for Inngest

Django Inngest

Missing Django integration for Inngest. This package depends on the Inngest Python SDK (pypi)

Installation

uv add inngest django-inngest

Configuration

Update urls.py to include:

urlpatterns = [
    # ...
    path("api/inngest/", include("django_inngest.urls")),
    # ...
]

Update settings.py to include:

# required for inngest and production
INNGEST_EVENT_KEY = os.environ.get("INNGEST_EVENT_KEY", None)
INNGEST_SIGNING_KEY = os.environ.get("INNGEST_SIGNING_KEY", None)

# configuration
DJANGO_INNGEST_APP_ID = os.environ.get("DJANGO_INNGEST_APP_ID", "my-django-project")
DJANGO_INNGEST_IS_PRODUCTION = not DEBUG
DJANGO_INNGEST_CLIENT_LOGGER = os.environ.get(
    "DJANGO_INNGEST_CLIENT_LOGGER", "gunicorn"
)
DJANGO_INNGEST_INACTIVE_FUNCTION_IDS = os.environ.get(
    "DJANGO_INNGEST_INACTIVE_FUNCTION_IDS", []
)
DJANGO_INNGEST_AUTO_DISCOVER_FUNCTIONS = os.environ.get(
    "DJANGO_INNGEST_AUTO_DISCOVER_FUNCTIONS", True
)

Usage

Create a new Django app and workflows file:

python manage.py startapp myapp
touch myapp/workflows.py

Update INSTALLED_APPS to include myapp.

In myapp/workflows.py, create a new Inngest function:

from datetime import timedelta

import inngest
from django_inngest.client import inngest_client


@inngest_client.create_function(
    id="myapp/my-function",
    trigger=inngest.TriggerEvent(name="my/trigger/event"),
    rate_limit=inngest.RateLimit(
        limit=1,
        period=timedelta(minutes=2, seconds=30),
        key="event.data.channel_id",
    ),
)
def my_function(ctx: inngest.Context):
    print(ctx.event.data.get("channel_id"))
    # your inngest function logic here
    return "OK"

Then trigger anywhere in your Django project with:

import inngest
from django_inngest.client import inngest_client

event = inngest.Event(
    name="my/trigger/event",
    data={"channel_id": "1234567890"},
)

inngest_client.send(event)