Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/comm/__init__.py: 72%

1"""Comm package. 

2 

3Copyright (c) IPython Development Team. 

4Distributed under the terms of the Modified BSD License. 

5 

6This package provides a way to register a Kernel Comm implementation, as per 

7the Jupyter kernel protocol. 

8It also provides a base Comm implementation and a default CommManager for the IPython case. 

9""" 

10from __future__ import annotations 

11 

12from typing import Any 

13 

14from .base_comm import BaseComm, BuffersType, CommManager, MaybeDict 

15 

16__version__ = "0.2.2" 

17__all__ = [ 

18 "create_comm", 

19 "get_comm_manager", 

20 "__version__", 

21] 

22 

23_comm_manager = None 

24 

25 

26class DummyComm(BaseComm): 

27 def publish_msg( 

28 self, 

29 msg_type: str, 

30 data: MaybeDict = None, 

31 metadata: MaybeDict = None, 

32 buffers: BuffersType = None, 

33 **keys: Any, 

34 ) -> None: 

35 pass 

36 

37 

38def _create_comm(*args: Any, **kwargs: Any) -> BaseComm: 

39 """Create a Comm. 

40 

41 This method is intended to be replaced, so that it returns your Comm instance. 

42 """ 

43 return DummyComm(*args, **kwargs) 

44 

45 

46def _get_comm_manager() -> CommManager: 

47 """Get the current Comm manager, creates one if there is none. 

48 

49 This method is intended to be replaced if needed (if you want to manage multiple CommManagers). 

50 """ 

51 global _comm_manager # noqa: PLW0603 

52 

53 if _comm_manager is None: 

54 _comm_manager = CommManager() 

55 

56 return _comm_manager 

57 

58 

59create_comm = _create_comm 

60get_comm_manager = _get_comm_manager