Coverage for /pythoncovmergedfiles/medio/medio/src/jupyter_server/jupyter_server/services/kernels/connection/abc.py: 75%

1from abc import ABC, abstractmethod 

2from typing import Any 

3 

4 

5class KernelWebsocketConnectionABC(ABC): 

6 """ 

7 This class defines a minimal interface that should 

8 be used to bridge the connection between Jupyter 

9 Server's websocket API and a kernel's ZMQ socket 

10 interface. 

11 """ 

12 

13 websocket_handler: Any 

14 

15 @abstractmethod 

16 async def connect(self): 

17 """Connect the kernel websocket to the kernel ZMQ connections""" 

18 ... 

19 

20 @abstractmethod 

21 async def disconnect(self): 

22 """Disconnect the kernel websocket from the kernel ZMQ connections""" 

23 ... 

24 

25 @abstractmethod 

26 def handle_incoming_message(self, incoming_msg: str) -> None: 

27 """Broker the incoming websocket message to the appropriate ZMQ channel.""" 

28 ... 

29 

30 @abstractmethod 

31 def handle_outgoing_message(self, stream: str, outgoing_msg: list) -> None: 

32 """Broker outgoing ZMQ messages to the kernel websocket.""" 

33 ...