From 3ef14d5860eb20b24811ac555c4fb90d89ab81db Mon Sep 17 00:00:00 2001 From: Dan Farrelly Date: Tue, 16 Sep 2025 18:56:18 -0400 Subject: [PATCH 1/5] Inject run_id into Step class This metadata is required to publish realtime messages. --- pkg/inngest/inngest/_internal/comm_lib/handler.py | 2 ++ pkg/inngest/inngest/_internal/step_lib/base.py | 2 ++ pkg/inngest/inngest/_internal/step_lib/step_async.py | 2 ++ pkg/inngest/inngest/_internal/step_lib/step_sync.py | 2 ++ pkg/inngest/inngest/experimental/mocked/trigger.py | 2 ++ 5 files changed, 10 insertions(+) diff --git a/pkg/inngest/inngest/_internal/comm_lib/handler.py b/pkg/inngest/inngest/_internal/comm_lib/handler.py index 0fcecaff..a0689155 100644 --- a/pkg/inngest/inngest/_internal/comm_lib/handler.py +++ b/pkg/inngest/inngest/_internal/comm_lib/handler.py @@ -185,6 +185,7 @@ async def post( middleware, step_lib.StepIDCounter(), params.step_id, + request.ctx.run_id, ), ), params.fn_id, @@ -224,6 +225,7 @@ async def post( middleware, step_lib.StepIDCounter(), params.step_id, + request.ctx.run_id, ), ), params.fn_id, diff --git a/pkg/inngest/inngest/_internal/step_lib/base.py b/pkg/inngest/inngest/_internal/step_lib/base.py index 29102c95..33d3ec8b 100644 --- a/pkg/inngest/inngest/_internal/step_lib/base.py +++ b/pkg/inngest/inngest/_internal/step_lib/base.py @@ -87,11 +87,13 @@ def __init__( middleware: middleware_lib.MiddlewareManager, step_id_counter: StepIDCounter, target_hashed_id: typing.Optional[str], + run_id: str, ) -> None: self._client = client self._middleware = middleware self._step_id_counter = step_id_counter self._target_hashed_id = target_hashed_id + self._run_id = run_id def _handle_skip( self, diff --git a/pkg/inngest/inngest/_internal/step_lib/step_async.py b/pkg/inngest/inngest/_internal/step_lib/step_async.py index 3c9ee9fa..d12c01fb 100644 --- a/pkg/inngest/inngest/_internal/step_lib/step_async.py +++ b/pkg/inngest/inngest/_internal/step_lib/step_async.py @@ -31,12 +31,14 @@ def __init__( middleware: middleware_lib.MiddlewareManager, step_id_counter: base.StepIDCounter, target_hashed_id: typing.Optional[str], + run_id: str, ) -> None: super().__init__( client, middleware, step_id_counter, target_hashed_id, + run_id, ) self.ai = AI(self) diff --git a/pkg/inngest/inngest/_internal/step_lib/step_sync.py b/pkg/inngest/inngest/_internal/step_lib/step_sync.py index a572d6f6..675c77e7 100644 --- a/pkg/inngest/inngest/_internal/step_lib/step_sync.py +++ b/pkg/inngest/inngest/_internal/step_lib/step_sync.py @@ -29,12 +29,14 @@ def __init__( middleware: middleware_lib.MiddlewareManager, step_id_counter: base.StepIDCounter, target_hashed_id: typing.Optional[str], + run_id: str, ) -> None: super().__init__( client, middleware, step_id_counter, target_hashed_id, + run_id, ) self.ai = AI(self) diff --git a/pkg/inngest/inngest/experimental/mocked/trigger.py b/pkg/inngest/inngest/experimental/mocked/trigger.py index fe91fb7c..172988ad 100644 --- a/pkg/inngest/inngest/experimental/mocked/trigger.py +++ b/pkg/inngest/inngest/experimental/mocked/trigger.py @@ -99,6 +99,7 @@ def trigger( middleware, step_lib.StepIDCounter(), step_id, + request.ctx.run_id, ), ) @@ -133,6 +134,7 @@ def trigger( middleware, step_lib.StepIDCounter(), step_id, + request.ctx.run_id, ), ) From 38d86d0ec37c18cdcb06b6a0764560a93d9a56fb Mon Sep 17 00:00:00 2001 From: Dan Farrelly Date: Tue, 16 Sep 2025 18:58:54 -0400 Subject: [PATCH 2/5] Add post request utility method --- .../inngest/_internal/client_lib/client.py | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/pkg/inngest/inngest/_internal/client_lib/client.py b/pkg/inngest/inngest/_internal/client_lib/client.py index 75406820..dcecdc7e 100644 --- a/pkg/inngest/inngest/_internal/client_lib/client.py +++ b/pkg/inngest/inngest/_internal/client_lib/client.py @@ -359,6 +359,39 @@ def _get_sync(self, url: str) -> types.MaybeError[httpx.Response]: return res + async def _post( + self, url: str, body: object + ) -> types.MaybeError[httpx.Response]: + """ + Perform an asynchronous HTTP POST request. Handles authn + """ + req = self._http_client.build_request( + "POST", + url, + headers=net.create_headers( + env=self._env, + framework=None, + server_kind=None, + ), + json=body, + timeout=self._httpx_timeout, + ) + + res = await net.fetch_with_auth_fallback( + self._http_client, + self._http_client_sync, + req, + signing_key=self._signing_key, + signing_key_fallback=self._signing_key_fallback, + ) + if isinstance(res, Exception): + return res + + if res.status_code >= 400: + return Exception(f"HTTP error: {res.status_code} {res.text}") + + return res + async def _get_batch( self, run_id: str ) -> types.MaybeError[list[server_lib.Event]]: From 37437698d8a9cabf8246b7576198ec1653a5e79b Mon Sep 17 00:00:00 2001 From: Dan Farrelly Date: Tue, 16 Sep 2025 18:59:47 -0400 Subject: [PATCH 3/5] Add experimental publish utility --- .../inngest/experimental/realtime/__init__.py | 9 ++++ .../inngest/experimental/realtime/publish.py | 44 +++++++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 pkg/inngest/inngest/experimental/realtime/__init__.py create mode 100644 pkg/inngest/inngest/experimental/realtime/publish.py diff --git a/pkg/inngest/inngest/experimental/realtime/__init__.py b/pkg/inngest/inngest/experimental/realtime/__init__.py new file mode 100644 index 00000000..dbfe72d2 --- /dev/null +++ b/pkg/inngest/inngest/experimental/realtime/__init__.py @@ -0,0 +1,9 @@ +""" +A helper library for realtime publishing + +This is an experimental preview support for the Python SDK. +""" + +from .publish import publish + +__all__ = ["publish"] diff --git a/pkg/inngest/inngest/experimental/realtime/publish.py b/pkg/inngest/inngest/experimental/realtime/publish.py new file mode 100644 index 00000000..35b91603 --- /dev/null +++ b/pkg/inngest/inngest/experimental/realtime/publish.py @@ -0,0 +1,44 @@ +import json +import typing +from urllib.parse import urlencode, urljoin + +from inngest._internal import errors, step_lib + + +# Q - Should this be step_publish or realtime publish or otherwise to compare to publish within a step? +async def publish( + step: step_lib.Step, + channel: str, + topic: str, + data: typing.Mapping[str, object], +) -> None: + """ + Publish a message to a realtime channel. + This currently requires the Step object as an argument as the API is finalized. + """ + + params = { + "channel": channel, + "topic": topic, + "run_id": step._run_id, + } + + async def _publish_api_request() -> None: + res = await step._client._post( + url=urljoin( + step._client._api_origin, + f"/v1/realtime/publish?{urlencode(params)}", + ), + body=data, + ) + if isinstance(res, Exception): + raise res + if res.status_code != 200: + raise errors.Error( + "failed to publish to realtime channel", + ) + return None + + await step.run(f"publish:{channel}", _publish_api_request) + + return None From fe1326829b34da747aaf498c9cd9d043edf47425 Mon Sep 17 00:00:00 2001 From: Dan Farrelly Date: Tue, 16 Sep 2025 19:56:35 -0400 Subject: [PATCH 4/5] Add get_subscription_token --- .../inngest/experimental/realtime/__init__.py | 3 +- .../inngest/experimental/realtime/publish.py | 1 + .../realtime/subscription_tokens.py | 39 +++++++++++++++++++ 3 files changed, 42 insertions(+), 1 deletion(-) create mode 100644 pkg/inngest/inngest/experimental/realtime/subscription_tokens.py diff --git a/pkg/inngest/inngest/experimental/realtime/__init__.py b/pkg/inngest/inngest/experimental/realtime/__init__.py index dbfe72d2..00329584 100644 --- a/pkg/inngest/inngest/experimental/realtime/__init__.py +++ b/pkg/inngest/inngest/experimental/realtime/__init__.py @@ -5,5 +5,6 @@ """ from .publish import publish +from .subscription_tokens import get_subscription_token -__all__ = ["publish"] +__all__ = ["publish", "get_subscription_token"] diff --git a/pkg/inngest/inngest/experimental/realtime/publish.py b/pkg/inngest/inngest/experimental/realtime/publish.py index 35b91603..9871d28b 100644 --- a/pkg/inngest/inngest/experimental/realtime/publish.py +++ b/pkg/inngest/inngest/experimental/realtime/publish.py @@ -6,6 +6,7 @@ # Q - Should this be step_publish or realtime publish or otherwise to compare to publish within a step? +# TODO - Support streams async def publish( step: step_lib.Step, channel: str, diff --git a/pkg/inngest/inngest/experimental/realtime/subscription_tokens.py b/pkg/inngest/inngest/experimental/realtime/subscription_tokens.py new file mode 100644 index 00000000..f49e93fe --- /dev/null +++ b/pkg/inngest/inngest/experimental/realtime/subscription_tokens.py @@ -0,0 +1,39 @@ +import typing +from urllib.parse import urljoin + +from inngest._internal import client_lib, errors + + +async def get_subscription_token( + client: client_lib.Inngest, channel: str, topics: list[str] +) -> typing.Mapping[str, object]: + """ + Create a subscription token for a given channel and topics. + The token can be used by a client to subscribe to realtime events, + including front-end applications using the @inngest/realtime npm package. + """ + data = [] + for topic in topics: + data.append( + { + "channel": channel, + "topic": topic, + "kind": "run", + } + ) + + res = await client._post( + url=urljoin( + client._api_origin, + "/v1/realtime/token", + ), + body=data, + ) + if isinstance(res, Exception): + raise res + if res.status_code >= 300: + raise errors.Error( + "failed to get subscription token", + ) + # Response is an object with a "jwt" property which is a string + return res.json() From 79f56c227f4c667f6062070dd3648e52dee561c6 Mon Sep 17 00:00:00 2001 From: Dan Farrelly Date: Tue, 16 Sep 2025 20:05:03 -0400 Subject: [PATCH 5/5] Update app id in connect example --- examples/connect/src/inngest/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/connect/src/inngest/client.py b/examples/connect/src/inngest/client.py index 8632eb36..436decad 100644 --- a/examples/connect/src/inngest/client.py +++ b/examples/connect/src/inngest/client.py @@ -3,4 +3,4 @@ logger = structlog.get_logger() -inngest_client = inngest.Inngest(app_id="fast_api_example", logger=logger) +inngest_client = inngest.Inngest(app_id="connect_example", logger=logger)