Rename run_replayer, remove need to return self from init

Author: tconley1428

README.md

@@ -1516,9 +1516,8 @@ class AuthenticationPlugin(Plugin):
     def __init__(self, api_key: str):
         self.api_key = api_key
         
-    def init_client_plugin(self, next: Plugin) -> Plugin:
+    def init_client_plugin(self, next: Plugin) -> None:
       self.next_client_plugin = next
-      return self
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         # Modify client configuration
@@ -1552,16 +1551,15 @@ Here's an example of a worker plugin that adds custom monitoring:
 import temporalio
 from contextlib import asynccontextmanager
 from typing import AsyncIterator
-from temporalio.worker import Plugin, WorkerConfig, Worker, ReplayerConfig, Worker, Replayer, WorkflowReplayResult
+from temporalio.worker import Plugin, WorkerConfig, ReplayerConfig, Worker, Replayer, WorkflowReplayResult
 import logging
 
 class MonitoringPlugin(Plugin):
     def __init__(self):
         self.logger = logging.getLogger(__name__)
 
-    def init_worker_plugin(self, next: Plugin) -> Plugin:
+    def init_worker_plugin(self, next: Plugin) -> None:
         self.next_worker_plugin = next
-        return self
 
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
         # Modify worker configuration
@@ -1581,14 +1579,14 @@ class MonitoringPlugin(Plugin):
       return self.next_worker_plugin.configure_replayer(config)
 
   @asynccontextmanager
-  async def workflow_replay(
+  async def run_replayer(
       self,
       replayer: Replayer,
       histories: AsyncIterator[temporalio.client.WorkflowHistory],
   ) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]:
         self.logger.info("Starting replay execution")
         try:
-          async with self.next_worker_plugin.workflow_replay(replayer, histories) as results:
+          async with self.next_worker_plugin.run_replayer(replayer, histories) as results:
               yield results
         finally:
             self.logger.info("Replay execution completed")
@@ -1614,13 +1612,11 @@ from temporalio.worker import Plugin as WorkerPlugin, WorkerConfig, ReplayerConf
 
 
 class UnifiedPlugin(ClientPlugin, WorkerPlugin):
-    def init_client_plugin(self, next: ClientPlugin) -> ClientPlugin:
+    def init_client_plugin(self, next: ClientPlugin) -> None:
         self.next_client_plugin = next
-        return self
 
-    def init_worker_plugin(self, next: WorkerPlugin) -> WorkerPlugin:
+    def init_worker_plugin(self, next: WorkerPlugin) -> None:
         self.next_worker_plugin = next
-        return self
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         # Client-side customization
@@ -1646,12 +1642,12 @@ class UnifiedPlugin(ClientPlugin, WorkerPlugin):
         config["data_converter"] = pydantic_data_converter
         return config
 
-    async def workflow_replay(
+    async def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[temporalio.client.WorkflowHistory],
     ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
-        return self.next_worker_plugin.workflow_replay(replayer, histories)
+        return self.next_worker_plugin.run_replayer(replayer, histories)
 
 # Create client with the unified plugin
 client = await Client.connect(

temporalio/client.py

@@ -191,7 +191,8 @@ async def connect(
 
         root_plugin: Plugin = _RootPlugin()
         for plugin in reversed(plugins):
-            root_plugin = plugin.init_client_plugin(root_plugin)
+            plugin.init_client_plugin(root_plugin)
+            root_plugin = plugin
 
         service_client = await root_plugin.connect_service_client(connect_config)
 
@@ -235,7 +236,8 @@ def __init__(
 
         root_plugin: Plugin = _RootPlugin()
         for plugin in reversed(plugins):
-            root_plugin = plugin.init_client_plugin(root_plugin)
+            plugin.init_client_plugin(root_plugin)
+            root_plugin = plugin
 
         self._init_from_config(root_plugin.configure_client(config))
 
@@ -7399,7 +7401,7 @@ def name(self) -> str:
         return type(self).__module__ + "." + type(self).__qualname__
 
     @abstractmethod
-    def init_client_plugin(self, next: Plugin) -> Plugin:
+    def init_client_plugin(self, next: Plugin) -> None:
         """Initialize this plugin in the plugin chain.
 
         This method sets up the chain of responsibility pattern by storing a reference
@@ -7408,9 +7410,6 @@ def init_client_plugin(self, next: Plugin) -> Plugin:
 
         Args:
             next: The next plugin in the chain to delegate to.
-
-        Returns:
-            This plugin instance for method chaining.
         """
 
     @abstractmethod
@@ -7447,7 +7446,7 @@ async def connect_service_client(
 
 
 class _RootPlugin(Plugin):
-    def init_client_plugin(self, next: Plugin) -> Plugin:
+    def init_client_plugin(self, next: Plugin) -> None:
         raise NotImplementedError()
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:

temporalio/contrib/openai_agents/_temporal_openai_agents.py

@@ -237,25 +237,19 @@ def __init__(
         self._model_params = model_params
         self._model_provider = model_provider
 
-    def init_client_plugin(
-        self, next: temporalio.client.Plugin
-    ) -> temporalio.client.Plugin:
+    def init_client_plugin(self, next: temporalio.client.Plugin) -> None:
         """Set the next client plugin"""
         self.next_client_plugin = next
-        return self
 
     async def connect_service_client(
         self, config: temporalio.service.ConnectConfig
     ) -> temporalio.service.ServiceClient:
         """No modifications to service client"""
         return await self.next_client_plugin.connect_service_client(config)
 
-    def init_worker_plugin(
-        self, next: temporalio.worker.Plugin
-    ) -> temporalio.worker.Plugin:
+    def init_worker_plugin(self, next: temporalio.worker.Plugin) -> None:
         """Set the next worker plugin"""
         self.next_worker_plugin = next
-        return self
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         """Configure the Temporal client for OpenAI agents integration.
@@ -320,14 +314,14 @@ def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
         return config
 
     @asynccontextmanager
-    async def workflow_replay(
+    async def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[temporalio.client.WorkflowHistory],
     ) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]:
         """Set the OpenAI Overrides during replay"""
         with set_open_ai_agent_temporal_overrides(self._model_params):
-            async with self.next_worker_plugin.workflow_replay(
+            async with self.next_worker_plugin.run_replayer(
                 replayer, histories
             ) as results:
                 yield results

temporalio/worker/_plugin.py

@@ -35,7 +35,7 @@ def name(self) -> str:
         return type(self).__module__ + "." + type(self).__qualname__
 
     @abc.abstractmethod
-    def init_worker_plugin(self, next: Plugin) -> Plugin:
+    def init_worker_plugin(self, next: Plugin) -> None:
         """Initialize this plugin in the plugin chain.
 
         This method sets up the chain of responsibility pattern by storing a reference
@@ -44,9 +44,6 @@ def init_worker_plugin(self, next: Plugin) -> Plugin:
 
         Args:
             next: The next plugin in the chain to delegate to.
-
-        Returns:
-            This plugin instance for method chaining.
         """
 
     @abc.abstractmethod
@@ -92,7 +89,7 @@ def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
         """
 
     @abc.abstractmethod
-    def workflow_replay(
+    def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[WorkflowHistory],
@@ -101,7 +98,7 @@ def workflow_replay(
 
 
 class _RootPlugin(Plugin):
-    def init_worker_plugin(self, next: Plugin) -> Plugin:
+    def init_worker_plugin(self, next: Plugin) -> None:
         raise NotImplementedError()
 
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
@@ -113,7 +110,7 @@ def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
     async def run_worker(self, worker: Worker) -> None:
         await worker._run()
 
-    def workflow_replay(
+    def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[WorkflowHistory],

temporalio/worker/_replayer.py

@@ -86,7 +86,8 @@ def __init__(
         # Apply plugin configuration
         root_plugin: temporalio.worker.Plugin = _RootPlugin()
         for plugin in reversed(plugins):
-            root_plugin = plugin.init_worker_plugin(root_plugin)
+            plugin.init_worker_plugin(root_plugin)
+            root_plugin = plugin
         self._config = root_plugin.configure_replayer(self._config)
         self._plugin = root_plugin
 
@@ -175,7 +176,7 @@ def workflow_replay_iterator(
             An async iterator that returns replayed workflow results as they are
             replayed.
         """
-        return self._plugin.workflow_replay(self, histories)
+        return self._plugin.run_replayer(self, histories)
 
     @asynccontextmanager
     async def _workflow_replay_iterator(

temporalio/worker/_worker.py

@@ -374,7 +374,8 @@ def __init__(
 
         root_plugin: Plugin = _RootPlugin()
         for plugin in reversed(plugins):
-            root_plugin = plugin.init_worker_plugin(root_plugin)
+            plugin.init_worker_plugin(root_plugin)
+            root_plugin = plugin
         config = root_plugin.configure_worker(config)
         self._plugin = root_plugin
 

tests/test_plugins.py

@@ -37,9 +37,8 @@ class MyClientPlugin(temporalio.client.Plugin):
     def __init__(self):
         self.interceptor = TestClientInterceptor()
 
-    def init_client_plugin(self, next: Plugin) -> Plugin:
+    def init_client_plugin(self, next: Plugin) -> None:
         self.next_client_plugin = next
-        return self
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         config["namespace"] = "replaced_namespace"
@@ -74,17 +73,11 @@ async def test_client_plugin(client: Client, env: WorkflowEnvironment):
 
 
 class MyCombinedPlugin(temporalio.client.Plugin, temporalio.worker.Plugin):
-    def init_worker_plugin(
-        self, next: temporalio.worker.Plugin
-    ) -> temporalio.worker.Plugin:
+    def init_worker_plugin(self, next: temporalio.worker.Plugin) -> None:
         self.next_worker_plugin = next
-        return self
 
-    def init_client_plugin(
-        self, next: temporalio.client.Plugin
-    ) -> temporalio.client.Plugin:
+    def init_client_plugin(self, next: temporalio.client.Plugin) -> None:
         self.next_client_plugin = next
-        return self
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         return self.next_client_plugin.configure_client(config)
@@ -104,20 +97,17 @@ async def run_worker(self, worker: Worker) -> None:
     def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
         return self.next_worker_plugin.configure_replayer(config)
 
-    def workflow_replay(
+    def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[temporalio.client.WorkflowHistory],
     ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
-        return self.next_worker_plugin.workflow_replay(replayer, histories)
+        return self.next_worker_plugin.run_replayer(replayer, histories)
 
 
 class MyWorkerPlugin(temporalio.worker.Plugin):
-    def init_worker_plugin(
-        self, next: temporalio.worker.Plugin
-    ) -> temporalio.worker.Plugin:
+    def init_worker_plugin(self, next: temporalio.worker.Plugin) -> None:
         self.next_worker_plugin = next
-        return self
 
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
         config["task_queue"] = "replaced_queue"
@@ -135,12 +125,12 @@ async def run_worker(self, worker: Worker) -> None:
     def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
         return self.next_worker_plugin.configure_replayer(config)
 
-    def workflow_replay(
+    def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[temporalio.client.WorkflowHistory],
     ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
-        return self.next_worker_plugin.workflow_replay(replayer, histories)
+        return self.next_worker_plugin.run_replayer(replayer, histories)
 
 
 async def test_worker_plugin_basic_config(client: Client) -> None:
@@ -203,17 +193,11 @@ async def test_worker_sandbox_restrictions(client: Client) -> None:
 
 
 class ReplayCheckPlugin(temporalio.client.Plugin, temporalio.worker.Plugin):
-    def init_worker_plugin(
-        self, next: temporalio.worker.Plugin
-    ) -> temporalio.worker.Plugin:
+    def init_worker_plugin(self, next: temporalio.worker.Plugin) -> None:
         self.next_worker_plugin = next
-        return self
 
-    def init_client_plugin(
-        self, next: temporalio.client.Plugin
-    ) -> temporalio.client.Plugin:
+    def init_client_plugin(self, next: temporalio.client.Plugin) -> None:
         self.next_client_plugin = next
-        return self
 
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         config["data_converter"] = pydantic_data_converter
@@ -237,14 +221,12 @@ async def connect_service_client(
         return await self.next_client_plugin.connect_service_client(config)
 
     @asynccontextmanager
-    async def workflow_replay(
+    async def run_replayer(
         self,
         replayer: Replayer,
         histories: AsyncIterator[temporalio.client.WorkflowHistory],
     ) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]:
-        async with self.next_worker_plugin.workflow_replay(
-            replayer, histories
-        ) as result:
+        async with self.next_worker_plugin.run_replayer(replayer, histories) as result:
             yield result