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

1"""Base class for a Comm""" 

2 

3# Copyright (c) IPython Development Team. 

4# Distributed under the terms of the Modified BSD License. 

5 

6import uuid 

7from typing import Optional 

8from warnings import warn 

9 

10import comm.base_comm 

11import traitlets.config 

12from traitlets import Bool, Bytes, Instance, Unicode, default 

13 

14from ipykernel.jsonutil import json_clean 

15from ipykernel.kernelbase import Kernel 

16 

17 

18# this is the class that will be created if we do comm.create_comm 

19class BaseComm(comm.base_comm.BaseComm): # type:ignore[misc] 

20 """The base class for comms.""" 

21 

22 kernel: Optional["Kernel"] = None 

23 

24 def publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys): 

25 """Helper for sending a comm message on IOPub""" 

26 if not Kernel.initialized(): 

27 return 

28 

29 data = {} if data is None else data 

30 metadata = {} if metadata is None else metadata 

31 content = json_clean(dict(data=data, comm_id=self.comm_id, **keys)) 

32 

33 if self.kernel is None: 

34 self.kernel = Kernel.instance() 

35 

36 assert self.kernel.session is not None 

37 self.kernel.session.send( 

38 self.kernel.iopub_socket, 

39 msg_type, 

40 content, 

41 metadata=json_clean(metadata), 

42 parent=self.kernel.get_parent(), 

43 ident=self.topic, 

44 buffers=buffers, 

45 ) 

46 

47 

48# but for backwards compatibility, we need to inherit from LoggingConfigurable 

49class Comm(BaseComm, traitlets.config.LoggingConfigurable): 

50 """Class for communicating between a Frontend and a Kernel""" 

51 

52 kernel = Instance("ipykernel.kernelbase.Kernel", allow_none=True) # type:ignore[assignment] 

53 comm_id = Unicode() 

54 primary = Bool(True, help="Am I the primary or secondary Comm?") 

55 

56 target_name = Unicode("comm") 

57 target_module = Unicode( 

58 None, 

59 allow_none=True, 

60 help="""requirejs module from 

61 which to load comm target.""", 

62 ) 

63 

64 topic = Bytes() 

65 

66 @default("kernel") 

67 def _default_kernel(self): 

68 if Kernel.initialized(): 

69 return Kernel.instance() 

70 return None 

71 

72 @default("comm_id") 

73 def _default_comm_id(self): 

74 return uuid.uuid4().hex 

75 

76 def __init__( 

77 self, target_name="", data=None, metadata=None, buffers=None, show_warning=True, **kwargs 

78 ): 

79 """Initialize a comm.""" 

80 if show_warning: 

81 warn( 

82 "The `ipykernel.comm.Comm` class has been deprecated. Please use the `comm` module instead." 

83 "For creating comms, use the function `from comm import create_comm`.", 

84 DeprecationWarning, 

85 stacklevel=2, 

86 ) 

87 

88 # Handle differing arguments between base classes. 

89 had_kernel = "kernel" in kwargs 

90 kernel = kwargs.pop("kernel", None) 

91 if target_name: 

92 kwargs["target_name"] = target_name 

93 BaseComm.__init__(self, data=data, metadata=metadata, buffers=buffers, **kwargs) # type:ignore[call-arg] 

94 # only re-add kernel if explicitly provided 

95 if had_kernel: 

96 kwargs["kernel"] = kernel 

97 traitlets.config.LoggingConfigurable.__init__(self, **kwargs) 

98 

99 

100__all__ = ["Comm"]