inngest 0.1.0a4

Python SDK for Inngest

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.

---

Inngest Python SDK

🚧 Currently in alpha! Not guaranteed to be production ready! 🚧

Supported frameworks:

• Fast API
• Flask
• Tornado

Usage

• Basic
• Step run
• Avoiding async/await

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.create_function(
    fn_id="find_person",
    trigger=inngest.TriggerEvent(event="app/person.find"),
)
async def fetch_person(
    *,
    event: inngest.Event,
    step: inngest.Step,
    **_kwargs: object,
) -> dict:
    person_id = event.data["person_id"]
    res = requests.get(f"https://swapi.dev/api/people/{person_id}")
    return res.json()


app = flask.Flask(__name__)
inngest_client = inngest.Inngest(app_id="flask_example")

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

app.run(port=8000)

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.create_function(
    fn_id="find_ships",
    trigger=inngest.TriggerEvent(event="app/ships.find"),
)
async def fetch_ships(
    *,
    event: inngest.Event,
    step: inngest.Step,
    **_kwargs: object,
) -> dict:
    """
    Find all the ships a person has.
    """

    person_id = 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", lambda: _fetch_ship(ship_url))

        ship_names.append(ship["name"])

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

Avoiding async functions

Completely avoiding `async`` functions only requires 2 differences:

1. Use step: inngest.StepSync instead of step: inngest.Step
2. Use serve_sync instead of serve

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


app = flask.Flask(__name__)
inngest_client = inngest.Inngest(app_id="flask_example")

inngest.flask.serve_sync(
    app,
    inngest_client,
    [fetch_person],
)

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