Add step.wait_for_event

Makefile

@@ -16,7 +16,7 @@ install: check-venv
 	@pip install '.[extra]' -c constraints.txt
 
 itest: check-venv
-	@pytest tests
+	@pytest -n 4 -v tests
 
 pre-commit: format-check lint type-check utest
 
@@ -27,4 +27,4 @@ type-check: check-venv
 	@mypy inngest tests
 
 utest: check-venv
-	@pytest inngest
+	@pytest -v inngest

examples/functions/__init__.py

@@ -5,6 +5,7 @@
     print_event,
     send_event,
     two_steps_and_sleep,
+    wait_for_event,
 )
 
 functions = [
@@ -14,6 +15,7 @@
     print_event.fn,
     send_event.fn,
     two_steps_and_sleep.fn,
+    wait_for_event.fn,
 ]
 
 __all__ = ["functions"]

examples/functions/wait_for_event.py

@@ -0,0 +1,14 @@
+import inngest
+
+
+@inngest.create_function(
+    inngest.FunctionOpts(id="wait_for_event", name="wait_for_event"),
+    inngest.TriggerEvent(event="app/wait_for_event"),
+)
+def fn(*, step: inngest.Step, **_kwargs: object) -> None:
+    res = step.wait_for_event(
+        "wait",
+        event="app/wait_for_event.fulfill",
+        timeout=inngest.Duration.second(2),
+    )
+    step.run("print-result", lambda: print(res))

inngest/__init__.py

@@ -1,4 +1,5 @@
 from ._internal.client import Inngest
+from ._internal.const import Duration
 from ._internal.errors import NonRetriableError
 from ._internal.event import Event
 from ._internal.frameworks import flask, tornado
@@ -14,6 +15,7 @@
 __all__ = [
     "BatchConfig",
     "CancelConfig",
+    "Duration",
     "Event",
     "Function",
     "FunctionOpts",

inngest/_internal/comm.py

@@ -20,7 +20,7 @@
 from .errors import (
     InternalError,
     InvalidBaseURL,
-    InvalidFunctionConfig,
+    InvalidConfig,
     MissingFunction,
 )
 from .execution import Call, CallError
@@ -154,7 +154,7 @@ def call_function(
     def get_function_configs(self, app_url: str) -> list[FunctionConfig]:
         configs = [fn.get_config(app_url) for fn in self._fns.values()]
         if len(configs) == 0:
-            raise InvalidFunctionConfig("no functions found")
+            raise InvalidConfig("no functions found")
         return configs
 
     def _parse_registration_response(

inngest/_internal/comm_test.py

@@ -8,7 +8,7 @@
 import inngest
 
 from .comm import CommHandler
-from .errors import InvalidFunctionConfig
+from .errors import InvalidConfig
 
 
 class Test_get_function_configs(TestCase):  # pylint: disable=invalid-name
@@ -29,7 +29,7 @@ def test_full_config(self) -> None:
                 batch_events=inngest.BatchConfig(max_size=2, timeout="1m"),
                 cancel=inngest.CancelConfig(
                     event="app/cancel",
-                    if_expression="true",
+                    if_exp="true",
                     timeout="1m",
                 ),
                 id="fn",
@@ -60,5 +60,5 @@ def test_no_functions(self) -> None:
             logger=self.client.logger,
         )
 
-        with pytest.raises(InvalidFunctionConfig, match="no functions found"):
+        with pytest.raises(InvalidConfig, match="no functions found"):
             comm.get_function_configs("http://foo.bar")

inngest/_internal/const.py

@@ -8,6 +8,28 @@
 VERSION: Final = "0.1.0"
 
 
+class Duration:
+    @classmethod
+    def second(cls, count: int = 1) -> int:
+        return count * 60 * 1000
+
+    @classmethod
+    def minute(cls, count: int = 1) -> int:
+        return count * cls.second(60)
+
+    @classmethod
+    def hour(cls, count: int = 1) -> int:
+        return count * cls.minute(60)
+
+    @classmethod
+    def day(cls, count: int = 1) -> int:
+        return count * cls.hour(24)
+
+    @classmethod
+    def week(cls, count: int = 1) -> int:
+        return count * cls.day(7)
+
+
 class EnvKey(Enum):
     BASE_URL = "INNGEST_BASE_URL"
     EVENT_KEY = "INNGEST_EVENT_KEY"

inngest/_internal/errors.py

@@ -24,7 +24,7 @@ def __init__(self, message: str | None = None) -> None:
         )
 
 
-class InvalidFunctionConfig(InternalError):
+class InvalidConfig(InternalError):
     status_code: int = 500
 
     def __init__(self, message: str | None = None) -> None:
@@ -37,7 +37,7 @@ def __init__(self, message: str | None = None) -> None:
     def from_validation_error(
         cls,
         err: ValidationError,
-    ) -> InvalidFunctionConfig:
+    ) -> InvalidConfig:
         """
         Extract info from Pydantic's ValidationError and return our internal
         InvalidFunctionConfig error.

inngest/_internal/execution.py

@@ -37,8 +37,10 @@ class CallResponse(BaseModel):
     id: str
     name: str
     op: Opcode
+    opts: dict[str, object] | None = None
 
 
 class Opcode(Enum):
     SLEEP = "Sleep"
     STEP = "Step"
+    WAIT_FOR_EVENT = "WaitForEvent"

inngest/_internal/execution_test.py

@@ -8,6 +8,7 @@ def test_serialization() -> None:
         id="my_id",
         name="my_name",
         op=Opcode.STEP,
+        opts={},
     ).to_dict()
 
     expectation = {
@@ -16,6 +17,7 @@ def test_serialization() -> None:
         "id": "my_id",
         "name": "my_name",
         "op": "Step",
+        "opts": {},
     }
 
     assert actual == expectation

inngest/_internal/function.py

@@ -9,11 +9,7 @@
 from pydantic import ValidationError
 
 from .client import Inngest
-from .errors import (
-    InvalidFunctionConfig,
-    NonRetriableError,
-    UnserializableOutput,
-)
+from .errors import InvalidConfig, NonRetriableError, UnserializableOutput
 from .event import Event
 from .execution import Call, CallError, CallResponse, Opcode
 from .function_config import (
@@ -27,7 +23,7 @@
     TriggerCron,
     TriggerEvent,
 )
-from .transforms import hash_step_id, to_iso_utc
+from .transforms import hash_step_id, to_duration_str, to_iso_utc
 from .types import BaseModel, EmptySentinel, T
 
 
@@ -53,7 +49,7 @@ def convert_validation_error(
         self,
         err: ValidationError,
     ) -> BaseException:
-        return InvalidFunctionConfig.from_validation_error(err)
+        return InvalidConfig.from_validation_error(err)
 
 
 class Function:
@@ -86,6 +82,7 @@ def call(
                     id=out.hashed_id,
                     name=out.name,
                     op=out.op,
+                    opts=out.opts,
                 )
             ]
         except Exception as err:
@@ -146,12 +143,14 @@ def __init__(
         hashed_id: str,
         name: str,
         op: Opcode,
+        opts: dict[str, object] | None = None,
     ) -> None:
         self.data = data
         self.display_name = display_name
         self.hashed_id = hashed_id
         self.name = name
         self.op = op
+        self.opts = opts
 
 
 class _Step:
@@ -228,6 +227,49 @@ def sleep_until(
             op=Opcode.SLEEP,
         )
 
+    def wait_for_event(
+        self,
+        id: str,  # pylint: disable=redefined-builtin
+        *,
+        event: str,
+        if_exp: str | None = None,
+        timeout: int,
+    ) -> Event | None:
+        """
+        Args:
+            event: Event name.
+            if_exp: An expression to filter events.
+            timeout: The maximum number of milliseconds to wait for the event.
+        """
+
+        id_count = self._step_id_counter.increment(id)
+        if id_count > 1:
+            id = f"{id}:{id_count - 1}"
+        hashed_id = hash_step_id(id)
+
+        memo = self._get_memo(hashed_id)
+        if memo is not EmptySentinel:
+            if memo is None:
+                # Timeout
+                return None
+
+            # Fulfilled by an event
+            return Event.model_validate(memo)
+
+        opts: dict[str, object] = {
+            "timeout": to_duration_str(timeout),
+        }
+        if if_exp is not None:
+            opts["if"] = if_exp
+
+        raise EarlyReturn(
+            hashed_id=hashed_id,
+            display_name=id,
+            name=event,
+            op=Opcode.WAIT_FOR_EVENT,
+            opts=opts,
+        )
+
 
 class _FunctionHandler(Protocol):
     def __call__(self, *, event: Event, step: Step) -> object:
@@ -256,6 +298,16 @@ def sleep_until(
     ) -> None:
         ...
 
+    def wait_for_event(
+        self,
+        id: str,  # pylint: disable=redefined-builtin
+        *,
+        event: str,
+        if_exp: str | None = None,
+        timeout: int,
+    ) -> Event | None:
+        ...
+
 
 class _StepIDCounter:
     def __init__(self) -> None:

inngest/_internal/function_config.py

@@ -4,7 +4,7 @@
 
 from pydantic import Field, ValidationError
 
-from .errors import InvalidFunctionConfig
+from .errors import InvalidConfig
 from .types import BaseModel
 
 # A number > 0 followed by a time unit (s, m, h, d, w)
@@ -16,12 +16,12 @@ def convert_validation_error(
         self,
         err: ValidationError,
     ) -> BaseException:
-        return InvalidFunctionConfig.from_validation_error(err)
+        return InvalidConfig.from_validation_error(err)
 
 
 class CancelConfig(_BaseConfig):
     event: str
-    if_expression: str | None = None
+    if_exp: str | None = None
     timeout: str | None = Field(default=None, pattern=TIME_PERIOD_REGEX)
 
 

inngest/_internal/transforms.py

@@ -2,6 +2,8 @@
 import re
 from datetime import datetime, timezone
 
+from .const import Duration
+from .errors import InvalidConfig
 from .types import T
 
 
@@ -37,3 +39,18 @@ def to_iso_utc(value: datetime) -> str:
         value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
         + "Z"
     )
+
+
+def to_duration_str(ms: int) -> str:
+    if ms < Duration.second():
+        raise InvalidConfig("duration must be at least 1 second")
+    if ms < Duration.minute():
+        return f"{ms // Duration.second()}s"
+    if ms < Duration.hour():
+        return f"{ms // Duration.minute()}m"
+    if ms < Duration.day():
+        return f"{ms // Duration.hour()}h"
+    if ms < Duration.week():
+        return f"{ms // Duration.day()}d"
+
+    return f"{ms // Duration.week()}w"

pyproject.toml

@@ -21,6 +21,7 @@ extra = [
     "mypy==1.6.1",
     "pylint==3.0.1",
     "pytest==7.4.2",
+    "pytest-xdist[psutil]==3.3.1",
     "python-json-logger==2.0.7",
     "toml==0.10.2",
     "tornado==6.3.3",

tests/base.py

@@ -1,5 +1,4 @@
-import time
-from typing import Callable, Protocol
+from typing import Protocol
 
 import requests
 
@@ -36,20 +35,3 @@ def set_up(case: _FrameworkTestCase) -> None:
 
 def tear_down(case: _FrameworkTestCase) -> None:
     case.http_proxy.stop()
-
-
-def wait_for(
-    assertion: Callable[[], None],
-    timeout: int = 5,
-) -> None:
-    start = time.time()
-    while True:
-        try:
-            assertion()
-            return
-        except Exception as err:
-            timed_out = time.time() - start > timeout
-            if timed_out:
-                raise err
-
-        time.sleep(0.2)