Continue to support Kernel._parent_ident for backward compatibility

ipykernel/kernelbase.py

@@ -16,6 +16,7 @@
 import typing as t
 import uuid
 import warnings
+from collections.abc import Mapping
 from contextvars import ContextVar
 from datetime import datetime
 from functools import partial
@@ -61,6 +62,7 @@
 
 from ._version import kernel_protocol_version
 from .iostream import OutStream
+from .utils import LazyDict
 
 _AWAITABLE_MESSAGE: str = (
     "For consistency across implementations, it is recommended that `{func_name}`"
@@ -199,6 +201,9 @@ def _default_ident(self):
     _control_parent_ident: bytes = b""
     _shell_parent: ContextVar[dict[str, Any]]
     _shell_parent_ident: ContextVar[bytes]
+    # Kept for backward-compatibility, accesses _control_parent_ident and _shell_parent_ident,
+    # see https://github.com/jupyterlab/jupyterlab/issues/17785
+    _parent_ident: Mapping[str, bytes]
 
     @property
     def _parent_header(self):
@@ -313,6 +318,15 @@ def __init__(self, **kwargs):
         self._shell_parent_ident = ContextVar("shell_parent_ident")
         self._shell_parent_ident.set(b"")
 
+        # For backward compatibility so that _parent_ident["shell"] and _parent_ident["control"]
+        # work as they used to for ipykernel >= 7
+        self._parent_ident = LazyDict(
+            {
+                "control": lambda: self._control_parent_ident,
+                "shell": lambda: self._shell_parent_ident.get(),
+            }
+        )
+
     async def dispatch_control(self, msg):
         """Dispatch a control request, ensuring only one message is processed at a time."""
         # Ensure only one control message is processed at a time

ipykernel/utils.py

@@ -0,0 +1,26 @@
+"""Utilities"""
+
+import typing as t
+from collections.abc import Mapping
+
+
+class LazyDict(Mapping[str, t.Any]):
+    """Lazy evaluated read-only dictionary.
+
+    Initialised with a dictionary of key-value pairs where the values are either
+    constants or callables. Callables are evaluated each time the respective item is
+    read.
+    """
+
+    def __init__(self, dict):
+        self._dict = dict
+
+    def __getitem__(self, key):
+        item = self._dict.get(key)
+        return item() if callable(item) else item
+
+    def __len__(self):
+        return len(self._dict)
+
+    def __iter__(self):
+        return iter(self._dict)

pyproject.toml

@@ -88,7 +88,7 @@ path = "ipykernel/_version.py"
 features = ["docs"]
 [tool.hatch.envs.docs.scripts]
 build = "make -C docs html SPHINXOPTS='-W'"
-api = "sphinx-apidoc -o docs/api -f -E ipykernel tests ipykernel/datapub.py ipykernel/pickleutil.py ipykernel/serialize.py ipykernel/gui ipykernel/pylab"
+api = "sphinx-apidoc -o docs/api -f -E ipykernel tests ipykernel/datapub.py ipykernel/pickleutil.py ipykernel/serialize.py ipykernel/gui ipykernel/pylab ipykernel/utils.py"
 
 [tool.hatch.envs.test]
 features = ["test"]

tests/test_kernel.py

@@ -717,3 +717,44 @@ def test_shutdown_subprocesses():
             child_newpg.terminate()
         except psutil.NoSuchProcess:
             pass
+
+
+def test_parent_header_and_ident():
+    # Kernel._parent_ident is private but kept for backward compatibility,
+    # see https://github.com/jupyterlab/jupyterlab/issues/17785
+    with kernel() as kc:
+        # get_parent('shell')
+        msg_id, _ = execute(
+            kc=kc,
+            code="k=get_ipython().kernel; p=k.get_parent('shell'); print(p['header']['msg_id'], p['header']['session'])",
+        )
+        stdout, _ = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_id)
+        check_msg_id, session = stdout.split()
+        assert check_msg_id == msg_id
+        assert check_msg_id.startswith(msg_id)
+
+        # _parent_ident['shell']
+        msg_id, _ = execute(kc=kc, code="print(k._parent_ident['shell'])")
+        stdout, _ = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_id)
+        assert stdout == f"[b'{session}']\n"
+
+        # Send a control message
+        msg = kc.session.msg("kernel_info_request")
+        kc.control_channel.send(msg)
+        control_msg_id = msg["header"]["msg_id"]
+        assemble_output(kc.get_iopub_msg, parent_msg_id=control_msg_id)
+
+        # get_parent('control')
+        msg_id, _ = execute(
+            kc=kc,
+            code="p=k.get_parent('control'); print(p['header']['msg_id'], p['header']['session'])",
+        )
+        stdout, _ = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_id)
+        check_msg_id, session = stdout.split()
+        assert check_msg_id == control_msg_id
+        assert check_msg_id.startswith(control_msg_id)
+
+        # _parent_ident['control']
+        msg_id, _ = execute(kc=kc, code="print(k._parent_ident['control'])")
+        stdout, _ = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_id)
+        assert stdout == f"[b'{session}']\n"