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]