Update pipeline runner signature (#429)

Remove default value for `ctx_id`;
Fix type annotations for `ctx_id`;
Convert `ctx_id` to string in websocket tutorial.

Author: RLKRo

chatsky/core/pipeline.py

@@ -233,7 +233,7 @@ def validate_fallback_label(self):
         return self
 
     async def _run_pipeline(
-        self, request: Message, ctx_id: Optional[str] = None, update_ctx_misc: Optional[dict] = None
+        self, request: Message, ctx_id: Optional[str], update_ctx_misc: Optional[dict] = None
     ) -> Context:
         """
         Method that should be invoked on user input.
@@ -292,9 +292,7 @@ def run(self):
         logger.info("Pipeline is accepting requests.")
         asyncio.run(self.messenger_interface.connect(self._run_pipeline))
 
-    def __call__(
-        self, request: Message, ctx_id: Optional[str] = None, update_ctx_misc: Optional[dict] = None
-    ) -> Context:
+    def __call__(self, request: Message, ctx_id: Optional[str], update_ctx_misc: Optional[dict] = None) -> Context:
         """
         Method that executes pipeline once.
         Basically, it is a shortcut for :py:meth:`_run_pipeline`.

chatsky/core/service/types.py

@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 from enum import unique, Enum
-from typing import Callable, Union, Awaitable, Optional, Any, Protocol, Hashable, TYPE_CHECKING
+from typing import Callable, Union, Awaitable, Optional, Any, Protocol, TYPE_CHECKING
 from typing_extensions import TypeAlias
 from pydantic import BaseModel
 
@@ -25,13 +25,12 @@ class PipelineRunnerFunction(Protocol):
     """
 
     def __call__(
-        self, message: Message, ctx_id: Optional[Hashable] = None, update_ctx_misc: Optional[dict] = None
+        self, message: Message, ctx_id: Optional[str], update_ctx_misc: Optional[dict] = None
     ) -> Awaitable[Context]:
         """
         :param message: User request for pipeline to process.
         :param ctx_id:
             ID of the context that the new request belongs to.
-            Optional, None by default.
             If set to `None`, a new context will be created with `message` being the first request.
         :param update_ctx_misc:
             Dictionary to be passed as an argument to `ctx.misc.update`.

chatsky/messengers/common/interface.py

@@ -11,7 +11,7 @@
 import logging
 from pathlib import Path
 from tempfile import gettempdir
-from typing import Optional, Any, List, Tuple, Hashable, TYPE_CHECKING, Type
+from typing import Optional, Any, List, Tuple, TYPE_CHECKING, Type
 
 if TYPE_CHECKING:
     from chatsky.core import Context
@@ -99,7 +99,7 @@ class PollingMessengerInterface(MessengerInterface):
     """
 
     @abc.abstractmethod
-    def _request(self) -> List[Tuple[Message, Hashable]]:
+    def _request(self) -> List[Tuple[Message, str]]:
         """
         Method used for sending users request for their input.
 
@@ -181,17 +181,15 @@ async def connect(self, pipeline_runner: PipelineRunnerFunction):
         self._pipeline_runner = pipeline_runner
 
     async def on_request_async(
-        self, request: Message, ctx_id: Optional[Hashable] = None, update_ctx_misc: Optional[dict] = None
+        self, request: Message, ctx_id: Optional[str], update_ctx_misc: Optional[dict] = None
     ) -> Context:
         """
         Method that should be invoked on user input.
         This method has the same signature as :py:class:`~chatsky.core.service.types.PipelineRunnerFunction`.
         """
         return await self._pipeline_runner(request, ctx_id, update_ctx_misc)
 
-    def on_request(
-        self, request: Any, ctx_id: Optional[Hashable] = None, update_ctx_misc: Optional[dict] = None
-    ) -> Context:
+    def on_request(self, request: Any, ctx_id: Optional[str], update_ctx_misc: Optional[dict] = None) -> Context:
         """
         Method that should be invoked on user input.
         This method has the same signature as :py:class:`~chatsky.core.service.types.PipelineRunnerFunction`.

chatsky/messengers/console.py

@@ -1,4 +1,4 @@
-from typing import Any, Hashable, List, Optional, TextIO, Tuple
+from typing import Any, List, Optional, TextIO, Tuple
 from uuid import uuid4
 from chatsky.messengers.common.interface import PollingMessengerInterface
 from chatsky.core.service.types import PipelineRunnerFunction
@@ -23,7 +23,7 @@ def __init__(
         out_descriptor: Optional[TextIO] = None,
     ):
         super().__init__()
-        self._ctx_id: Optional[Hashable] = None
+        self._ctx_id: Optional[str] = None
         self._intro: Optional[str] = intro
         self._prompt_request: str = prompt_request
         self._prompt_response: str = prompt_response

tests/core/test_actor.py

@@ -31,7 +31,7 @@ async def test_normal_execution(self):
             parallelize_processing=True,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.labels._items == {
             0: AbsoluteNodeLabel(flow_name="flow", node_name="node1"),
@@ -49,7 +49,7 @@ async def test_fallback_node(self):
             parallelize_processing=True,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.labels._items == {
             0: AbsoluteNodeLabel(flow_name="flow", node_name="node"),
@@ -85,7 +85,7 @@ async def test_default_priority(self, default_priority, result):
             default_priority=default_priority,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.last_label.node_name == result
 
@@ -105,7 +105,7 @@ async def call(self, ctx: Context) -> None:
             parallelize_processing=True,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.last_label.node_name == "fallback"
         assert log_list[0].msg == "Exception occurred during transition processing."
@@ -122,7 +122,7 @@ async def test_empty_response(self, log_event_catcher):
             parallelize_processing=True,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.responses == {1: Message()}
         assert log_list[-1].msg == "Node has empty response."
@@ -142,7 +142,7 @@ async def call(self, ctx: Context) -> MessageInitTypes:
             parallelize_processing=True,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.responses == {1: Message()}
         assert log_list[-1].msg == "Response was not produced."
@@ -162,7 +162,7 @@ async def call(self, ctx: Context) -> None:
             parallelize_processing=True,
         )
 
-        ctx = await pipeline._run_pipeline(Message())
+        ctx = await pipeline._run_pipeline(Message(), ctx_id="0")
 
         assert ctx.responses == {1: Message()}
         assert log_list[0].msg == "Exception occurred during response processing."

tests/core/test_context.py

@@ -125,7 +125,7 @@ async def call(self, ctx: Context) -> MessageInitTypes:
             return ctx.pipeline.start_label.node_name
 
     pipeline = Pipeline(script={"flow": {"node": {RESPONSE: MyResponse()}}}, start_label=("flow", "node"))
-    ctx = await pipeline._run_pipeline(Message(text=""))
+    ctx = await pipeline._run_pipeline(Message(text=""), ctx_id="0")
 
     assert ctx.last_response == Message(text="node")
 
@@ -145,7 +145,7 @@ async def call(self, ctx: Context) -> None:
         script={"flow": {"node": {PRE_RESPONSE: {"": MyProcessing()}, PRE_TRANSITION: {"": MyProcessing()}}}},
         start_label=("flow", "node"),
     )
-    ctx = await pipeline._run_pipeline(Message(text=""))
+    ctx = await pipeline._run_pipeline(Message(text=""), ctx_id="0")
     assert len(log) == 2
 
     ctx.framework_data.current_node = None

tests/core/test_message.py

@@ -2,7 +2,7 @@
 from pathlib import Path
 from random import randint
 from shutil import rmtree
-from typing import Hashable, Optional, TextIO
+from typing import Optional, TextIO
 from urllib.request import urlopen
 
 import pytest
@@ -48,7 +48,7 @@ class ChatskyCLIMessengerInterface(CLIMessengerInterface, MessengerInterfaceWith
 
     def __init__(self, attachments_directory: Optional[Path] = None):
         MessengerInterfaceWithAttachments.__init__(self, attachments_directory)
-        self._ctx_id: Optional[Hashable] = None
+        self._ctx_id: Optional[str] = None
         self._intro: Optional[str] = None
         self._prompt_request: str = "request: "
         self._prompt_response: str = "response: "

tests/messengers/telegram/utils.py

@@ -2,7 +2,7 @@
 from contextlib import contextmanager
 from importlib import import_module
 from hashlib import sha256
-from typing import Any, Dict, Hashable, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 from pydantic import BaseModel
 from telegram import InputFile, InputMedia, Update
@@ -90,7 +90,7 @@ def _wrap_pipeline_runner(self) -> Iterator[None]:
         original_pipeline_runner = self.interface._pipeline_runner
 
         async def wrapped_pipeline_runner(
-            message: Message, ctx_id: Optional[Hashable] = None, update_ctx_misc: Optional[dict] = None
+            message: Message, ctx_id: Optional[str], update_ctx_misc: Optional[dict] = None
         ) -> Context:
             self.latest_ctx = await original_pipeline_runner(message, ctx_id, update_ctx_misc)
             return self.latest_ctx

tests/pipeline/test_service.py

@@ -34,7 +34,7 @@ async def call(self, ctx: Context):
         pre_services=[MyProcessing(wait=0.02, text="A")],
         post_services=[MyProcessing(wait=0, text="C")],
     )
-    await pipeline._run_pipeline(Message(""))
+    await pipeline._run_pipeline(Message(""), ctx_id="0")
     assert logs == ["A", "B", "C"]
 
 

tutorials/messengers/web_api_interface/2_websocket_chat.py

@@ -94,7 +94,7 @@ async def websocket_endpoint(websocket: WebSocket, client_id: int):
             await websocket.send_text(f"User: {data}")
             request = Message(data)
             context = await messenger_interface.on_request_async(
-                request, client_id
+                request, str(client_id)
             )
             response = context.last_response.text
             if response is not None: