Skip to main content

Python SDK for Inngest

Project description




Serverless event-driven queues, background jobs, and scheduled jobs for Python.
Works with any framework and platform.

Read the documentation and get started in minutes.

pypi versions discord twitter


Inngest Python SDK

We currently support the following frameworks (but adding a new framework is easy!):

  • DigitalOcean Functions
  • Django (>=4.2)
  • FastAPI (>=0.100.0)
  • Flask (>=2.3.0)
  • Tornado (>=6.3)

Python 3.9 is the minimum version we support.

Getting started

Install inngest in your project:

pip install inngest

Write a basic function and serve it (see the basic example for guidance).

Start the Dev Server (the local version of our cloud platform):

npx inngest-cli@latest dev

Browse to http://127.0.0.1:8288 and you should see your app! Visit our docs to read more about the Dev Server.

Examples

💡 You can mix async and non-async functions in the same app!

Basic (no steps)

This is a minimal example of an Inngest function. It's missing some of our features but it's a good starting point.

import flask
import inngest.flask
import requests

inngest_client = inngest.Inngest(
    app_id="flask_example",
    is_production=False,
)

@inngest_client.create_function(
    fn_id="find_person",
    trigger=inngest.TriggerEvent(event="app/person.find"),
)
def fetch_person(
    ctx: inngest.Context,
    step: inngest.StepSync,
) -> dict:
    person_id = ctx.event.data["person_id"]
    res = requests.get(f"https://swapi.dev/api/people/{person_id}")
    return res.json()

app = flask.Flask(__name__)

# Register functions with the Inngest server
inngest.flask.serve(
    app,
    inngest_client,
    [fetch_person],
)

app.run(port=8000)

Send the following event in the Dev Server UI and the fetch_person function will run:

{
  "name": "app/person.find",
  "data": {
    "person_id": 1
  }
}

Step run

The following example registers a function that will:

  1. Get the person ID from the event
  2. Fetch the person with that ID
  3. Fetch the person's ships
  4. Return a summary dict
@inngest_client.create_function(
    fn_id="find_ships",
    trigger=inngest.TriggerEvent(event="app/ships.find"),
)
def fetch_ships(
    ctx: inngest.Context,
    step: inngest.StepSync,
) -> dict:
    """
    Find all the ships a person has.
    """

    person_id = ctx.event.data["person_id"]

    def _fetch_person() -> dict:
        res = requests.get(f"https://swapi.dev/api/people/{person_id}")
        return res.json()

    # Wrap the function with step.run to enable retries
    person = step.run("fetch_person", _fetch_person)

    def _fetch_ship(url: str) -> dict:
        res = requests.get(url)
        return res.json()

    ship_names = []
    for ship_url in person["starships"]:
        # step.run works in loops!
        ship = step.run("fetch_ship", _fetch_ship, ship_url)

        ship_names.append(ship["name"])

    return {
        "person_name": person["name"],
        "ship_names": ship_names,
    }

Send the following event in the Dev Server UI and the fetch_person function will run:

{
  "name": "app/ships.find",
  "data": {
    "person_id": 1
  }
}

Async function

@inngest_client.create_function(
    fn_id="find_person",
    trigger=inngest.TriggerEvent(event="app/person.find"),
)
async def fetch_person(
    ctx: inngest.Context,
    step: inngest.Step,
) -> dict:
    person_id = ctx.event.data["person_id"]
    async with httpx.AsyncClient() as client:
        res = await client.get(f"https://swapi.dev/api/people/{person_id}")
        return res.json()

Sending an event outside a function

Sometimes you want to send an event from a normal, non-Inngest function. You can do that using the client:

inngest_client.send_sync(inngest.Event(name="app/test", data={"person_id": 1}))

If you prefer async then use the send method instead:

await inngest_client.send(inngest.Event(name="app/test", data={"person_id": 1}))

Using in production

The Dev Server is not used in production. Inngest Cloud is used instead.

The INNGEST_EVENT_KEY and INNGEST_SIGNING_KEY environment variables must be set. These secrets establish trust between Inngest Cloud and your app. We also use request signature verification to mitigate man-in-the-middle attacks. You can read more about environment variables in our docs.

Your Inngest client must be in production mode. This is typically done with an environment variable:

inngest_client = inngest.Inngest(
    app_id="my_app",
    is_production=os.getenv("INNGEST_DEV") is None,
)

Project details


Release history Release notifications | RSS feed

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

inngest-0.3.15.tar.gz (47.6 kB view details)

Uploaded Source

Built Distribution

inngest-0.3.15-py3-none-any.whl (61.9 kB view details)

Uploaded Python 3

File details

Details for the file inngest-0.3.15.tar.gz.

File metadata

  • Download URL: inngest-0.3.15.tar.gz
  • Upload date:
  • Size: 47.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/5.0.0 CPython/3.12.3

File hashes

Hashes for inngest-0.3.15.tar.gz
Algorithm Hash digest
SHA256 62c318791a7316213dd9ed644c2f82ae7338da899143109fe4f6432cf8b313cf
MD5 9829ff5738c8ef7549044fe0b27ba77c
BLAKE2b-256 198be767b4f824ccc4ea6b943c92a983efe3cca4e9fb89c27128792173173fb1

See more details on using hashes here.

File details

Details for the file inngest-0.3.15-py3-none-any.whl.

File metadata

  • Download URL: inngest-0.3.15-py3-none-any.whl
  • Upload date:
  • Size: 61.9 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/5.0.0 CPython/3.12.3

File hashes

Hashes for inngest-0.3.15-py3-none-any.whl
Algorithm Hash digest
SHA256 fc67f87f6a341e4a0f6582c3c562f659ef96c64ef9e7e600cf80895c67a7a56d
MD5 1a5ad7bde1e1f967ff0a37d9cdc86a96
BLAKE2b-256 dae72c52d90df13c4923be0f7b5426f83fc176cdcf3ecd29e2cd3a06390daf6d

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page