Try to pass cell id to executing kernel.

Is seem that few kernel want to make use of cell id for various
reasons.

One way to get the cell_id in the executing context is to make use of
init_metadata but it is marked for removal.

for example in [there](https://github.com/robots-from-jupyter/robotkernel/blob/19b28267406d1f8555ce73b96e7efb8af8417266/src/robotkernel/kernel.py#L196-L205)

Another is to start passing cell_id down.

This is technically a change of API as our consumer may not support more
parameters so we take a peak at the signature, and pass cell_id only if
it is :
- an explicit parameter,
- the function/method takes varkwargs.

We could also start to refactor to pass a metadata object with multiple
fields if this is the preferred route.

One questions is whether and how we want to deprecated not receiving
cell_id as a parameter.

Related to ipython/ipython#13579 and
subsequent PR on IPython side to support cell_id in progress.

Author: Carreau

ipykernel/inprocess/tests/test_kernel.py

@@ -13,6 +13,12 @@
 from ipykernel.inprocess.ipkernel import InProcessKernel
 from ipykernel.inprocess.manager import InProcessKernelManager
 from ipykernel.tests.utils import assemble_output
+from IPython.utils.io import capture_output
+
+from jupyter_client.session import Session
+from contextlib import contextmanager
+
+orig_msg = Session.msg
 
 
 def _init_asyncio_patch():
@@ -53,6 +59,26 @@ def _init_asyncio_patch():
                 asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())
 
 
+def _inject_cell_id(_self, *args, **kwargs):
+    """
+    This patch jupyter_client.session:Session.msg to add a cell_id to the return message metadata
+    """
+    assert isinstance(_self, Session)
+    res = orig_msg(_self, *args, **kwargs)
+    assert "cellId" not in res["metadata"]
+    res["metadata"]["cellId"] = "test_cell_id"
+    return res
+
+
+@contextmanager
+def patch_cell_id():
+    try:
+        Session.msg = _inject_cell_id
+        yield
+    finally:
+        Session.msg = orig_msg
+
+
 class InProcessKernelTestCase(unittest.TestCase):
     def setUp(self):
         _init_asyncio_patch()
@@ -62,6 +88,11 @@ def setUp(self):
         self.kc.start_channels()
         self.kc.wait_for_ready()
 
+    def test_with_cell_id(self):
+
+        with patch_cell_id():
+            self.kc.execute("1+1")
+
     def test_pylab(self):
         """Does %pylab work in the in-process kernel?"""
         _ = pytest.importorskip("matplotlib", reason="This test requires matplotlib")

ipykernel/ipkernel.py

@@ -18,6 +18,7 @@
 from .debugger import Debugger, _is_debugpy_available
 from .eventloops import _use_appnope
 from .kernelbase import Kernel as KernelBase
+from .kernelbase import _accepts_cell_id
 from .zmqshell import ZMQInteractiveShell
 
 try:
@@ -296,7 +297,14 @@ def set_sigint_result():
             signal.signal(signal.SIGINT, save_sigint)
 
     async def do_execute(
-        self, code, silent, store_history=True, user_expressions=None, allow_stdin=False
+        self,
+        code,
+        silent,
+        store_history=True,
+        user_expressions=None,
+        allow_stdin=False,
+        *,
+        cell_id,
     ):
         shell = self.shell  # we'll need this a lot here
 
@@ -306,6 +314,7 @@ async def do_execute(
         if hasattr(shell, "run_cell_async") and hasattr(shell, "should_run_async"):
             run_cell = shell.run_cell_async
             should_run_async = shell.should_run_async
+            with_cell_id = _accepts_cell_id(run_cell)
         else:
             should_run_async = lambda cell: False  # noqa
             # older IPython,
@@ -314,6 +323,7 @@ async def do_execute(
             async def run_cell(*args, **kwargs):
                 return shell.run_cell(*args, **kwargs)
 
+            with_cell_id = _accepts_cell_id(shell.run_cell)
         try:
 
             # default case: runner is asyncio and asyncio is already running
@@ -336,13 +346,24 @@ async def run_cell(*args, **kwargs):
                     preprocessing_exc_tuple=preprocessing_exc_tuple,
                 )
             ):
-                coro = run_cell(
-                    code,
-                    store_history=store_history,
-                    silent=silent,
-                    transformed_cell=transformed_cell,
-                    preprocessing_exc_tuple=preprocessing_exc_tuple,
-                )
+                if with_cell_id:
+                    coro = run_cell(
+                        code,
+                        store_history=store_history,
+                        silent=silent,
+                        transformed_cell=transformed_cell,
+                        preprocessing_exc_tuple=preprocessing_exc_tuple,
+                        cell_id=cell_id,
+                    )
+                else:
+                    coro = run_cell(
+                        code,
+                        store_history=store_history,
+                        silent=silent,
+                        transformed_cell=transformed_cell,
+                        preprocessing_exc_tuple=preprocessing_exc_tuple,
+                    )
+
                 coro_future = asyncio.ensure_future(coro)
 
                 with self._cancel_on_sigint(coro_future):
@@ -357,7 +378,15 @@ async def run_cell(*args, **kwargs):
                 # runner isn't already running,
                 # make synchronous call,
                 # letting shell dispatch to loop runners
-                res = shell.run_cell(code, store_history=store_history, silent=silent)
+                if with_cell_id:
+                    res = shell.run_cell(
+                        code,
+                        store_history=store_history,
+                        silent=silent,
+                        cell_id=cell_id,
+                    )
+                else:
+                    res = shell.run_cell(code, store_history=store_history, silent=silent)
         finally:
             self._restore_input()
 
@@ -388,7 +417,8 @@ async def run_cell(*args, **kwargs):
 
         if "traceback" in reply_content:
             self.log.info(
-                "Exception in execute request:\n%s", "\n".join(reply_content["traceback"])
+                "Exception in execute request:\n%s",
+                "\n".join(reply_content["traceback"]),
             )
 
         # At this point, we can tell whether the main code execution succeeded
@@ -623,6 +653,7 @@ def __init__(self, *args, **kwargs):
         import warnings
 
         warnings.warn(
-            "Kernel is a deprecated alias of ipykernel.ipkernel.IPythonKernel", DeprecationWarning
+            "Kernel is a deprecated alias of ipykernel.ipkernel.IPythonKernel",
+            DeprecationWarning,
         )
         super().__init__(*args, **kwargs)

ipykernel/kernelbase.py

@@ -58,6 +58,14 @@
 from ._version import kernel_protocol_version
 
 
+def _accepts_cell_id(meth):
+    parameters = inspect.signature(meth).parameters
+    cid_param = parameters.get("cell_id")
+    return (cid_param and cid_param.kind == cid_param.KEYWORD_ONLY) or any(
+        p.kind == p.VAR_KEYWORD for p in parameters.values()
+    )
+
+
 class Kernel(SingletonConfigurable):
 
     # ---------------------------------------------------------------------------
@@ -690,13 +698,26 @@ async def execute_request(self, stream, ident, parent):
             self.execution_count += 1
             self._publish_execute_input(code, parent, self.execution_count)
 
-        reply_content = self.do_execute(
-            code,
-            silent,
-            store_history,
-            user_expressions,
-            allow_stdin,
-        )
+        cell_id = (parent.get("metadata") or {}).get("cellId")
+
+        if _accepts_cell_id(self.do_execute):
+            reply_content = self.do_execute(
+                code,
+                silent,
+                store_history,
+                user_expressions,
+                allow_stdin,
+                cell_id=cell_id,
+            )
+        else:
+            reply_content = self.do_execute(
+                code,
+                silent,
+                store_history,
+                user_expressions,
+                allow_stdin,
+            )
+
         if inspect.isawaitable(reply_content):
             reply_content = await reply_content
 
@@ -714,7 +735,12 @@ async def execute_request(self, stream, ident, parent):
         metadata = self.finish_metadata(parent, metadata, reply_content)
 
         reply_msg = self.session.send(
-            stream, "execute_reply", reply_content, parent, metadata=metadata, ident=ident
+            stream,
+            "execute_reply",
+            reply_content,
+            parent,
+            metadata=metadata,
+            ident=ident,
         )
 
         self.log.debug("%s", reply_msg)