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 return filefind(app.connection_file, [".", app.connection_dir]) 

38 

39 

40def _find_connection_file(connection_file): 

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

42 

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

44 - Otherwise, call jupyter_client.find_connection_file 

45 """ 

46 if connection_file is None: 

47 # get connection file from current kernel 

48 return get_connection_file() 

49 return jupyter_client.find_connection_file(connection_file) 

50 

51 

52def get_connection_info( 

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

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

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

56 

57 Parameters 

58 ---------- 

59 connection_file : str [optional] 

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

61 IPython will search in the security directory. 

62 If run from IPython, 

63 

64 If unspecified, the connection file for the currently running 

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

66 

67 unpack : bool [default: False] 

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

69 of the file. 

70 

71 Returns 

72 ------- 

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

74 depending on `unpack`. 

75 """ 

76 cf = _find_connection_file(connection_file) 

77 

78 with open(cf) as f: 

79 info_str = f.read() 

80 

81 if unpack: 

82 info = json.loads(info_str) 

83 # ensure key is bytes: 

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

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

86 

87 return info_str 

88 

89 

90def connect_qtconsole( 

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

92) -> Popen[Any]: 

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

94 

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

96 local notebook. 

97 

98 Parameters 

99 ---------- 

100 connection_file : str [optional] 

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

102 IPython will search in the security directory. 

103 If run from IPython, 

104 

105 If unspecified, the connection file for the currently running 

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

107 

108 argv : list [optional] 

109 Any extra args to be passed to the console. 

110 

111 Returns 

112 ------- 

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

114 """ 

115 argv = [] if argv is None else argv 

116 

117 cf = _find_connection_file(connection_file) 

118 

119 cmd = ";".join(["from qtconsole import qtconsoleapp", "qtconsoleapp.main()"]) 

120 

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

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

123 # interrupting the kernel doesn't kill it. 

124 kwargs["start_new_session"] = True 

125 

126 return Popen( 

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

128 stdout=PIPE, 

129 stderr=PIPE, 

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

131 **kwargs, 

132 ) 

133 

134 

135__all__ = [ 

136 "connect_qtconsole", 

137 "get_connection_file", 

138 "get_connection_info", 

139 "write_connection_file", 

140]