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

1"""Connection file-related utilities for the kernel""" 

2 

3# Copyright (c) IPython Development Team. 

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

5from __future__ import annotations 

6 

7import json 

8import sys 

9from subprocess import PIPE, Popen 

10from typing import TYPE_CHECKING, Any 

11 

12import jupyter_client 

13from jupyter_client import write_connection_file 

14 

15if TYPE_CHECKING: 

16 from ipykernel.kernelapp import IPKernelApp 

17 

18 

19def get_connection_file(app: IPKernelApp | None = None) -> str: 

20 """Return the path to the connection file of an app 

21 

22 Parameters 

23 ---------- 

24 app : IPKernelApp instance [optional] 

25 If unspecified, the currently running app will be used 

26 """ 

27 from traitlets.utils import filefind 

28 

29 if app is None: 

30 from ipykernel.kernelapp import IPKernelApp 

31 

32 if not IPKernelApp.initialized(): 

33 msg = "app not specified, and not in a running Kernel" 

34 raise RuntimeError(msg) 

35 

36 app = IPKernelApp.instance() 

37 # The kernel's own file lives in connection_dir; the working directory is 

38 # only a fallback for a relative name that was passed on the command line. 

39 return filefind(app.connection_file, [app.connection_dir, "."]) 

40 

41 

42def _find_connection_file(connection_file): 

43 """Return the absolute path for a connection file 

44 

45 - If nothing specified, return current Kernel's connection file 

46 - Otherwise, call jupyter_client.find_connection_file 

47 """ 

48 if connection_file is None: 

49 # get connection file from current kernel 

50 return get_connection_file() 

51 return jupyter_client.find_connection_file(connection_file) 

52 

53 

54def get_connection_info( 

55 connection_file: str | None = None, unpack: bool = False 

56) -> str | dict[str, Any]: 

57 """Return the connection information for the current Kernel. 

58 

59 Parameters 

60 ---------- 

61 connection_file : str [optional] 

62 The connection file to be used. Can be given by absolute path, or 

63 IPython will search in the security directory. 

64 If run from IPython, 

65 

66 If unspecified, the connection file for the currently running 

67 IPython Kernel will be used, which is only allowed from inside a kernel. 

68 

69 unpack : bool [default: False] 

70 if True, return the unpacked dict, otherwise just the string contents 

71 of the file. 

72 

73 Returns 

74 ------- 

75 The connection dictionary of the current kernel, as string or dict, 

76 depending on `unpack`. 

77 """ 

78 cf = _find_connection_file(connection_file) 

79 

80 with open(cf) as f: 

81 info_str = f.read() 

82 

83 if unpack: 

84 info = json.loads(info_str) 

85 # ensure key is bytes: 

86 info["key"] = info.get("key", "").encode() 

87 return info # type:ignore[no-any-return] 

88 

89 return info_str 

90 

91 

92def connect_qtconsole( 

93 connection_file: str | None = None, argv: list[str] | None = None 

94) -> Popen[Any]: 

95 """Connect a qtconsole to the current kernel. 

96 

97 This is useful for connecting a second qtconsole to a kernel, or to a 

98 local notebook. 

99 

100 Parameters 

101 ---------- 

102 connection_file : str [optional] 

103 The connection file to be used. Can be given by absolute path, or 

104 IPython will search in the security directory. 

105 If run from IPython, 

106 

107 If unspecified, the connection file for the currently running 

108 IPython Kernel will be used, which is only allowed from inside a kernel. 

109 

110 argv : list [optional] 

111 Any extra args to be passed to the console. 

112 

113 Returns 

114 ------- 

115 :class:`subprocess.Popen` instance running the qtconsole frontend 

116 """ 

117 argv = [] if argv is None else argv 

118 

119 cf = _find_connection_file(connection_file) 

120 

121 cmd = "from qtconsole import qtconsoleapp; qtconsoleapp.main()" 

122 

123 kwargs: dict[str, Any] = {} 

124 # Launch the Qt console in a separate session & process group, so 

125 # interrupting the kernel doesn't kill it. 

126 kwargs["start_new_session"] = True 

127 

128 return Popen( 

129 [sys.executable, "-c", cmd, "--existing", cf, *argv], 

130 stdout=PIPE, 

131 stderr=PIPE, 

132 close_fds=(sys.platform != "win32"), 

133 **kwargs, 

134 ) 

135 

136 

137__all__ = [ 

138 "connect_qtconsole", 

139 "get_connection_file", 

140 "get_connection_info", 

141 "write_connection_file", 

142]