An object-based approach to define flows

[slep0v]

Hello everyone! I did a bit of experimenting with inngest via python sdk and the value the product provides is fantastic!
(Context: I have an ML/DS/AI background and I see a lot of potential here for orchestrating realtime/HITL AI pipelines with your app).

But when implementing functions to be orchestrated by inngest in python, I have a feeling that I write a JS code.
In the text below I'd like to share my personal opinion, why I feel that way, and what I implemented on top of your sdk to enhance my DX:

1. Each step execution becomes a bit of a boilerplate: I first need to define a closure of my initial task, then pass it to step.run. While I'd be happy to just call y=f(x).
2. If I want to keep my repo modularized, I need to create a function that would take inngest.client and define a @client.create_function decorated flow inside.

This makes my code feel cluttered and harder to navigate than I'm used to. That's why I followed a pretty standard pattern in modern ML sdk's (pytorch model definition, huggingface model defition) and built a POC base class to build pipelines on.

class BaseTaskFlow:
    def __init__(self, client: inngest.Inngest, flow_name: str, trigger_event_url: str):
        self.client = client
        self.flow_name = flow_name
        self.trigger_event_url = trigger_event_url

        self.ctx: Optional[inngest.Context] = None
        self.step: Optional[inngest.Step] = None

        self.flow_fn = self.register_flow_fn()  # To be passed into inngest_{framework}.serve method

    def get_ctx(self) -> inngest.Context:
        return self.ctx

    @staticmethod
    def task(name: Optional[str] = None):
        """ Wraps a function to be executed as an Inngest step further. """
        def decorator(func):
            async def wrapped(self, *args, **kwargs):
                task_name = name or func.__name__
                result = await self.step.run(task_name, lambda: func(self, *args, **kwargs))
                return result

            return wrapped

        return decorator

    def register_flow_fn(self):
        @self.client.create_function(
            fn_id=self.flow_name,
            trigger=inngest.TriggerEvent(event=self.trigger_event_url),
        )
        async def flow_fn(ctx: inngest.Context, step: inngest.Step):
            self.ctx = ctx
            self.step = step
            res = await self.forward()
            return res

        return flow_fn

    async def forward(self) -> Any:
        """To be implemented by subclasses."""
        raise NotImplementedError

And then my flow (Inngest function) definition becomes very short and clear by subclassing a base helper class.

class MyTaskFlow(BaseTaskFlow):
    def __init__(self, client: inngest.Inngest, flow_name: str, trigger_event_url: str):
        super().__init__(client, flow_name, trigger_event_url)

    @BaseTaskFlow.task(name="greet_task")
    async def greet_task(self, name: Optional[str]) -> str:
        return f"Hello {name or 'World'}"

    @BaseTaskFlow.task(name="combine_task")
    async def combine_task(self, greeting: str, extra: str) -> str:
        return f"{greeting}, {extra}"

    async def forward(self) -> Any:
        ctx = self.get_ctx()
        greet = await self.greet_task(ctx.event.data.get("name"))
        final_message = await self.combine_task(greet, "How are you?")
        return final_message

This allows me to define statically all tasks that I need beforehand and keep the flow execution function clean - which is perfect for me.

This of course comes with some cons like:

• You cant define tasks dynamically.
• Cannot reuse tasks from other flows.

But they are perfectly fine for me: The first constraint is good for building reliable real-time inference DAGs, and if I need to reuse something, I can always create util operators and components.

Would love to know your thoughts on such approach and whether it is possible to see it being natively supported by sdk!

[amh4r]

Thank you so much for this thoughtful post! This conversation isn't limited to Python: it's relevant for OOP languages in general. Since we're writing SDKs in other OOP languages (Java is in progress and C# will eventually happen), we should have a broader discussion about SDK ergonomics in traditionally OOP languages.

We might take a while before replying with something definitive since we need to have a deep, forward-looking discussion about this