V1 release candidate

examples/functions/parallel_steps.py

@@ -21,7 +21,7 @@ async def _step_1b() -> int:
             await asyncio.sleep(2)
             return 2
 
-        return await step.parallel(
+        return await ctx.group.parallel(
             (
                 lambda: step.run("1a", _step_1a),
                 lambda: step.run("1b", _step_1b),
@@ -48,7 +48,7 @@ def _step_1b() -> int:
             time.sleep(2)
             return 2
 
-        return step.parallel(
+        return ctx.group.parallel_sync(
             (
                 lambda: step.run("1a", _step_1a),
                 lambda: step.run("1b", _step_1b),

inngest/__init__.py

@@ -22,7 +22,7 @@
     TriggerCron,
     TriggerEvent,
 )
-from ._internal.step_lib import Step, StepMemos, StepSync
+from ._internal.step_lib import Group, Step, StepMemos, StepSync
 from ._internal.types import JSON
 
 __all__ = [
@@ -33,6 +33,7 @@
     "Debounce",
     "Event",
     "Function",
+    "Group",
     "Inngest",
     "JSON",
     "Middleware",

inngest/_internal/comm_lib/handler.py

@@ -176,6 +176,7 @@ async def post(
                 attempt=request.ctx.attempt,
                 event=request.event,
                 events=events,
+                group=step_lib.Group(),
                 logger=self._client.logger,
                 run_id=request.ctx.run_id,
             ),
@@ -259,6 +260,7 @@ def post_sync(
                 attempt=request.ctx.attempt,
                 event=request.event,
                 events=events,
+                group=step_lib.Group(),
                 logger=self._client.logger,
                 run_id=request.ctx.run_id,
             ),

inngest/_internal/execution_lib/base.py

@@ -21,7 +21,6 @@ class BaseExecution(typing.Protocol):
     async def report_step(
         self,
         step_info: step_lib.StepInfo,
-        inside_parallel: bool,
     ) -> ReportedStep:
         ...
 
@@ -44,7 +43,6 @@ class BaseExecutionSync(typing.Protocol):
     def report_step(
         self,
         step_info: step_lib.StepInfo,
-        inside_parallel: bool,
     ) -> ReportedStepSync:
         ...
 

inngest/_internal/execution_lib/experimental.py

@@ -50,7 +50,6 @@ def __init__(
     async def report_step(
         self,
         step_info: step_lib.StepInfo,
-        inside_parallel: bool,
     ) -> execution_lib.ReportedStep:
         step_signal = asyncio.Future[execution_lib.ReportedStep]()
         self._pending_steps[step_info.id] = execution_lib.ReportedStep(

inngest/_internal/execution_lib/models.py

@@ -64,6 +64,7 @@ class Context:
     attempt: int
     event: server_lib.Event
     events: list[server_lib.Event]
+    group: step_lib.Group
     logger: types.Logger
     run_id: str
 

inngest/_internal/execution_lib/v0.py

@@ -59,7 +59,6 @@ def _handle_skip(
     async def report_step(
         self,
         step_info: step_lib.StepInfo,
-        inside_parallel: bool,
     ) -> ReportedStep:
         step_signal = asyncio.Future[ReportedStep]()
 
@@ -87,7 +86,7 @@ async def report_step(
         self._handle_skip(step_info)
 
         is_targeting_enabled = self._target_hashed_id is not None
-        if inside_parallel and not is_targeting_enabled:
+        if step_lib.in_parallel.get() and not is_targeting_enabled:
             if step_info.op == server_lib.Opcode.STEP_RUN:
                 step_info.op = server_lib.Opcode.PLANNED
 
@@ -244,7 +243,6 @@ def _handle_skip(
     def report_step(
         self,
         step_info: step_lib.StepInfo,
-        inside_parallel: bool,
     ) -> ReportedStepSync:
         step = ReportedStepSync(step_info)
 
@@ -269,7 +267,7 @@ def report_step(
         self._handle_skip(step_info)
 
         is_targeting_enabled = self._target_hashed_id is not None
-        if inside_parallel and not is_targeting_enabled:
+        if step_lib.in_parallel.get() and not is_targeting_enabled:
             if step_info.op == server_lib.Opcode.STEP_RUN:
                 step_info.op = server_lib.Opcode.PLANNED
 

inngest/_internal/step_lib/__init__.py

@@ -7,10 +7,12 @@
     StepMemos,
     StepResponse,
 )
+from .group import Group, in_parallel
 from .step_async import Step
 from .step_sync import StepSync
 
 __all__ = [
+    "Group",
     "ParsedStepID",
     "ResponseInterrupt",
     "SkipInterrupt",
@@ -20,4 +22,5 @@
     "StepMemos",
     "StepResponse",
     "StepSync",
+    "in_parallel",
 ]

inngest/_internal/step_lib/base.py

@@ -91,7 +91,6 @@ def __init__(
         target_hashed_id: typing.Optional[str],
     ) -> None:
         self._client = client
-        self._inside_parallel = False
         self._memos = memos
         self._middleware = middleware
         self._step_id_counter = step_id_counter

inngest/_internal/step_lib/group.py

@@ -0,0 +1,75 @@
+import contextvars
+import typing
+
+from inngest._internal import types
+
+from .base import ResponseInterrupt, SkipInterrupt, StepResponse
+
+# Create a context variable to track if we're in a parallel group
+in_parallel = contextvars.ContextVar("in_parallel", default=False)
+
+
+class Group:
+    async def parallel(
+        self,
+        callables: tuple[typing.Callable[[], typing.Awaitable[types.T]], ...],
+    ) -> tuple[types.T, ...]:
+        """
+        Run multiple steps in parallel.
+
+        Args:
+        ----
+            callables: An arbitrary number of step callbacks to run. These are callables that contain the step (e.g. `lambda: step.run("my_step", my_step_fn)`.
+        """
+
+        token = in_parallel.set(True)
+        try:
+            outputs = tuple[types.T]()
+            responses: list[StepResponse] = []
+            for cb in callables:
+                try:
+                    output = await cb()
+                    outputs = (*outputs, output)
+                except ResponseInterrupt as interrupt:
+                    responses = [*responses, *interrupt.responses]
+                except SkipInterrupt:
+                    pass
+
+            if len(responses) > 0:
+                raise ResponseInterrupt(responses)
+
+            return outputs
+        finally:
+            in_parallel.reset(token)
+
+    def parallel_sync(
+        self,
+        callables: tuple[typing.Callable[[], types.T], ...],
+    ) -> tuple[types.T, ...]:
+        """
+        Run multiple steps in parallel.
+
+        Args:
+        ----
+            callables: An arbitrary number of step callbacks to run. These are callables that contain the step (e.g. `lambda: step.run("my_step", my_step_fn)`.
+        """
+
+        token = in_parallel.set(True)
+        try:
+            outputs = tuple[types.T]()
+            responses: list[StepResponse] = []
+            for cb in callables:
+                try:
+                    output = cb()
+                    outputs = (*outputs, output)
+                except ResponseInterrupt as interrupt:
+                    responses = [*responses, *interrupt.responses]
+                except SkipInterrupt:
+                    pass
+
+            if len(responses) > 0:
+                raise ResponseInterrupt(responses)
+
+            return outputs
+        finally:
+            in_parallel.reset(token)

inngest/_internal/step_lib/step_async.py

@@ -132,10 +132,7 @@ async def invoke_by_id(
             opts=opts,
         )
 
-        async with await self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        async with await self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:
@@ -145,37 +142,6 @@ async def invoke_by_id(
 
         raise Exception("unreachable")
 
-    async def parallel(
-        self,
-        callables: tuple[typing.Callable[[], typing.Awaitable[types.T]], ...],
-    ) -> tuple[types.T, ...]:
-        """
-        Run multiple steps in parallel.
-
-        Args:
-        ----
-            callables: An arbitrary number of step callbacks to run. These are callables that contain the step (e.g. `lambda: step.run("my_step", my_step_fn)`.
-        """
-
-        self._inside_parallel = True
-
-        outputs = tuple[types.T]()
-        responses: list[base.StepResponse] = []
-        for cb in callables:
-            try:
-                output = await cb()
-                outputs = (*outputs, output)
-            except base.ResponseInterrupt as interrupt:
-                responses = [*responses, *interrupt.responses]
-            except base.SkipInterrupt:
-                pass
-
-        if len(responses) > 0:
-            raise base.ResponseInterrupt(responses)
-
-        self._inside_parallel = False
-        return outputs
-
     @typing.overload
     async def run(
         self,
@@ -232,10 +198,7 @@ async def run(
             op=server_lib.Opcode.STEP_RUN,
         )
 
-        async with await self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        async with await self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:
@@ -364,10 +327,7 @@ async def sleep_until(
             op=server_lib.Opcode.SLEEP,
         )
 
-        async with await self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        async with await self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:
@@ -417,10 +377,7 @@ async def wait_for_event(
             opts=opts,
         )
 
-        async with await self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        async with await self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:

inngest/_internal/step_lib/step_sync.py

@@ -147,10 +147,7 @@ def invoke_by_id(
             opts=opts,
         )
 
-        with self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        with self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:
@@ -160,37 +157,6 @@ def invoke_by_id(
 
         raise Exception("unreachable")
 
-    def parallel(
-        self,
-        callables: tuple[typing.Callable[[], types.T], ...],
-    ) -> tuple[types.T, ...]:
-        """
-        Run multiple steps in parallel.
-
-        Args:
-        ----
-            callables: An arbitrary number of step callbacks to run. These are callables that contain the step (e.g. `lambda: step.run("my_step", my_step_fn)`.
-        """
-
-        self._inside_parallel = True
-
-        outputs = tuple[types.T]()
-        responses: list[base.StepResponse] = []
-        for cb in callables:
-            try:
-                output = cb()
-                outputs = (*outputs, output)
-            except base.ResponseInterrupt as interrupt:
-                responses = [*responses, *interrupt.responses]
-            except base.SkipInterrupt:
-                pass
-
-        if len(responses) > 0:
-            raise base.ResponseInterrupt(responses)
-
-        self._inside_parallel = False
-        return outputs
-
     def run(
         self,
         step_id: str,
@@ -219,10 +185,7 @@ def run(
             op=server_lib.Opcode.STEP_RUN,
         )
 
-        with self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        with self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:
@@ -347,10 +310,7 @@ def sleep_until(
             op=server_lib.Opcode.SLEEP,
         )
 
-        with self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        with self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None:
@@ -418,10 +378,7 @@ def wait_for_event(
             opts=opts,
         )
 
-        with self._execution.report_step(
-            step_info,
-            self._inside_parallel,
-        ) as step:
+        with self._execution.report_step(step_info) as step:
             if step.skip:
                 raise base.SkipInterrupt(parsed_step_id.user_facing)
             if step.error is not None: