API Reference

MetaKernel

metakernel.MetaKernel

Bases: Kernel

The base MetaKernel class.

Source code in metakernel/_metakernel.py

class MetaKernel(Kernel):
    """The base MetaKernel class."""

    app_name = "metakernel"
    identifier_regex = r"[^\d\W][\w\.]*"
    func_call_regex = r"([^\d\W][\w\.]*)\([^\)\()]*\Z"
    magic_prefixes = dict(magic="%", shell="!", help="?")
    help_suffix = "?"
    help_links = [
        {
            "text": "MetaKernel Magics",
            "url": "https://metakernel.readthedocs.io/en/latest/source/README.html",
        },
    ]
    language_info: dict[str, Any] = {
        # 'mimetype': 'text/x-python',
        # 'name': 'python',
        # ------ If different from 'language':
        # 'codemirror_mode': {
        #    "version": 2,
        #    "name": "ipython"
        # }
        # 'pygments_lexer': 'language',
        # 'version'       : "x.y.z",
        # 'file_extension': '.py',
        "help_links": help_links,
    }
    plot_settings: dict[str, Any] = Dict(dict(backend="inline")).tag(config=True)  # type: ignore[assignment]

    meta_kernel = None

    @classmethod
    def run_as_main(cls, *args: Any, **kwargs: Any) -> None:
        """Launch or install a metakernel.

        Modules implementing a metakernel subclass can use the following lines:

            if __name__ == '__main__':
                MetaKernelSubclass.run_as_main()
        """
        kwargs["app_name"] = cls.app_name
        MetaKernelApp.launch_instance(kernel_class=cls, *args, **kwargs)  # noqa: B026

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)  # type:ignore[no-untyped-call]
        if MetaKernel.meta_kernel is None:
            MetaKernel.meta_kernel = self
        if self.log is None:
            # This occurs if we call as a stand-alone kernel
            # (eg, not as a process)
            # FIXME: take care of input/output, eg StringIO
            #        make work without a session
            self.log = logging.Logger(".metakernel")  # type:ignore[unreachable]
        else:
            # Write has already been set
            try:
                sys.stdout.write = self.Write  # type: ignore[method-assign]
            except Exception:  # noqa: S110
                pass  # Can't change stdout
        self.redirect_to_log = False
        self.shell = None
        self.sticky_magics: OrderedDict[str, str] = OrderedDict()
        self._i: str | None = None
        self._ii: str | None = None
        self._iii: str | None = None
        self._: Any = None
        self.__: Any = None
        self.___: Any = None
        self.max_hist_cache = 1000
        self.hist_cache: list[str] = []
        kwargs = {"parent": self, "kernel": self}
        self.comm_manager = comm.get_comm_manager()
        # widgets have changed target name in 8.x, keeping for compatibility
        self.comm_manager.register_target(
            "ipython.widget", lazy_import_handle_comm_opened
        )

        # compatible with widgets 8.x
        if Widget is not None:
            widgets.register_comm_target()

        # Ensure comm objects created by this kernel have kernel=self set at
        # construction time. This allows ipywidgets.Output context manager to
        # reach kernel.get_parent() and correctly set msg_id (issue #217).
        try:
            _kernel_ref = self
            _base_create_comm = comm.create_comm

            def _create_comm_with_kernel(*args: Any, **kwargs: Any) -> Any:
                c = _base_create_comm(*args, **kwargs)
                if getattr(c, "kernel", None) is None:
                    c.kernel = _kernel_ref  # type: ignore[attr-defined]
                return c

            comm.create_comm = _create_comm_with_kernel
        except Exception:  # noqa: S110
            pass

        self.hist_file = get_history_file(self)
        self.parser = Parser(
            self.identifier_regex,
            self.func_call_regex,
            self.magic_prefixes,
            self.help_suffix,
        )
        comm_msg_types = ["comm_open", "comm_msg", "comm_close"]
        for msg_type in comm_msg_types:
            self.shell_handlers[msg_type] = getattr(self.comm_manager, msg_type)
        self._display_formatter = DisplayFormatter()  # pass kwargs?
        self.env: dict[str, Any] = {}
        self.magic_search_paths: list[str] = []
        self.magic_load_errors: list[tuple[str, str]] = []
        self.reload_magics()
        # provide a way to get the current instance
        self.set_variable("kernel", self)
        # Run command line filenames, if given:
        if self.parent is not None and self.parent.extra_args:
            level = self.log.level
            self.log.setLevel("INFO")
            self.redirect_to_log = True
            self.Write("Executing files...")
            for filename in self.parent.extra_args:
                self.Write(f"    {filename}...")
                try:
                    self.do_execute_file(filename)
                except Exception as exc:
                    self.log.info(f"    {exc}")
            self.Write("Executing files: done!")
            self.log.setLevel(level)
            self.redirect_to_log = False

    def makeSubkernel(self, kernel: MetaKernel) -> None:
        """
        Run this method in an IPython kernel to set
        this kernel's input/output settings.
        """
        from IPython.display import display

        shell = get_ipython()
        if shell:  # we are running under an IPython kernel
            self.session = shell.kernel.session  # type: ignore[attr-defined]
            self.Display = display  # type: ignore[method-assign]
            self.send_response = self._send_shell_response  # type: ignore[method-assign]
        else:
            self.session = kernel.session
            self.send_response = kernel.send_response  # type: ignore[method-assign]
            self.Display = kernel.Display  # type: ignore[method-assign]

    #####################################
    # Methods which provide kernel - specific behavior

    def set_variable(self, name: str, value: Any) -> None:
        """
        Set a variable to a Python-typed value.
        """
        pass

    def get_variable(self, name: str) -> Any:
        """
        Lookup a variable name and return a Python-typed value.
        """
        pass

    def repr(self, item: Any) -> str:
        """The repr of the kernel."""
        return repr(item)

    def get_usage(self) -> str:
        """Get the usage statement for the kernel."""
        return "This is a usage statement."

    def get_kernel_help_on(
        self, info: dict[str, Any], level: int = 0, none_on_fail: bool = False
    ) -> str | None:
        """Get help on an object.  Called by the help magic."""
        if none_on_fail:
            return None
        else:
            return "Sorry, no help is available on '{}'.".format(info["code"])

    def handle_plot_settings(self) -> None:
        """Handle the current plot settings"""
        pass

    def get_local_magics_dir(self) -> str:
        """
        Returns the path to local magics dir (eg ~/.ipython/metakernel/magics)
        """
        base = get_ipython_dir()
        return os.path.join(base, "metakernel", "magics")

    def get_completions(self, info: dict[str, Any]) -> list[str]:
        """
        Get completions from kernel based on info dict.
        """
        return []

    def do_execute_direct(self, code: str, silent: bool = False) -> Any:
        """
        Execute code in the kernel language.
        """
        pass

    def do_execute_file(self, filename: str) -> Any:
        """
        Default code for running a file. Just opens the file, and sends
        the text to do_execute_direct.
        """
        with open(filename) as f:
            return self.do_execute_direct("".join(f.readlines()))

    def do_execute_meta(self, code: str) -> Any:
        """
        Execute meta code in the kernel. This uses the execute infrastructure
        but allows JavaScript to talk directly to the kernel bypassing normal
        processing.

        When responding to the %%debug magic, the step and reset meta
        commands can answer with a string in the format:

        "highlight: [start_line, start_col, end_line, end_col]"

        for highlighting expressions in the frontend.
        """
        if code == "reset":
            raise Exception("This kernel does not implement this meta command")
        elif code == "stop":
            raise Exception("This kernel does not implement this meta command")
        elif code == "step":
            raise Exception("This kernel does not implement this meta command")
        elif code.startswith("inspect "):
            raise Exception("This kernel does not implement this meta command")
        else:
            raise Exception(f"Unknown meta command: '{code}'")

    def initialize_debug(self, code: str) -> str:
        """
        This function is used with the %%debug magic for highlighting
        lines of code, and for initializing debug functions.

        Return the empty string if highlighting is not supported.
        """
        # return "highlight: [%s, %s, %s, %s]" % (line1, col1, line2, col2)
        return ""

    def do_function_direct(self, function_name: str, arg: Any) -> Any:
        """
        Call a function in the kernel language with args (as a single item).
        """
        f = self.do_execute_direct(function_name)
        return f(arg)

    def restart_kernel(self) -> None:
        """Restart the kernel"""
        pass

    def _request_shutdown(self) -> None:
        """Send an ask_exit payload and schedule kernel shutdown.

        Called when do_execute_direct raises SystemExit so the kernel closes
        gracefully instead of waiting for the parent-process monitor to fire.
        """
        self.kernel_resp["payload"] = [{"source": "ask_exit", "keepkernel": False}]

        # Suppress the parent-poller warning: we are shutting down intentionally,
        # so whichever exit path fires first (ours or the poller's) is fine.
        # The poller logs via traitlets.log.get_logger(), which returns the app's
        # logger directly — filters must be added there, not on logging.root (root
        # filters are not re-applied to propagated records).
        class _SuppressParentExit(logging.Filter):
            def filter(self, record: logging.LogRecord) -> bool:
                return "Parent appears to have exited" not in record.getMessage()

        _f = _SuppressParentExit()
        self.log.addFilter(_f)
        logging.root.addFilter(_f)

        loop = asyncio.get_event_loop()

        async def _shutdown() -> None:
            await self.do_shutdown(False)
            os._exit(0)

        loop.call_later(0.1, lambda: loop.create_task(_shutdown()))

    ############################################
    # Implement base class methods

    async def do_execute(
        self,
        code: Any,
        silent: Any = False,
        store_history: Any = True,
        user_expressions: Any = None,
        allow_stdin: Any = False,
        *,
        cell_meta: Any = None,
        cell_id: Any = None,
    ) -> Any:
        """Handle code execution.

        https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute
        """
        # Set the ability for the kernel to get standard-in:
        self._allow_stdin = allow_stdin
        # Create a default response:
        self.kernel_resp = {
            "status": "ok",
            # The base class increments the execution count
            "execution_count": self.execution_count,
            "payload": [],
            "user_expressions": {},
        }

        # TODO: remove this when IPython fixes this
        # This happens at startup when the language is set to python
        if "_usage.page_guiref" in code:
            return self.kernel_resp

        if code and store_history:
            self.hist_cache.append(code.strip())

        if not code.strip():
            return self.kernel_resp

        info = self.parse_code(code)
        self.payload = []
        retval = None

        if info["magic"] and info["magic"]["name"] == "help":
            if info["magic"]["type"] == "line":
                level = 0
            else:
                level = 1
            text = self.get_help_on(code, level)
            if text:
                content = {
                    "start_line_number": 0,
                    "source": "page",
                }
                if isinstance(text, dict):
                    content["data"] = text  ## {mime-type: ..., mime-type:...}
                    self.log.debug(str(text))
                else:
                    content["data"] = {"text/plain": text}
                    self.log.debug(text)
                self.payload = [content]

        elif info["magic"] or self.sticky_magics:
            retval = None
            if self.sticky_magics:
                magics, code = _split_magics_code(code, self.magic_prefixes)
                code = magics + self._get_sticky_magics() + code
            stack = []
            # Handle magics:
            magic = None
            prefixes = (self.magic_prefixes["shell"], self.magic_prefixes["magic"])
            while code.startswith(prefixes):
                magic = self.get_magic(code)
                if magic is not None:
                    stack.append(magic)
                    code = str(magic.get_code())
                    # signal to exit, maybe error or no block
                    if not magic.evaluate:
                        break
                else:
                    break  # type:ignore[unreachable]
            # Execute code, if any:
            if (magic is None or magic.evaluate) and code.strip() != "":
                if code.startswith("~~META~~:"):
                    retval = self.do_execute_meta(code[9:].strip())
                else:
                    try:
                        retval = self.do_execute_direct(code)
                        if inspect.isawaitable(retval):
                            retval = await retval
                    except SystemExit:
                        self._request_shutdown()
                        return self.kernel_resp
                    except Exception as e:
                        retval = ExceptionWrapper(type(e).__name__, str(e), [])
            # Post-process magics:
            for magic in reversed(stack):
                retval = magic.post_process(retval)
        else:
            if code.startswith("~~META~~:"):
                retval = self.do_execute_meta(code[9:].strip())
            else:
                try:
                    retval = self.do_execute_direct(code)
                    if inspect.isawaitable(retval):
                        retval = await retval
                except SystemExit:
                    self._request_shutdown()
                    return self.kernel_resp
                except Exception as e:
                    retval = ExceptionWrapper(type(e).__name__, str(e), [])

        await self.post_execute(retval, code, silent)

        if "payload" in self.kernel_resp:
            self.kernel_resp["payload"] = self.payload

        return self.kernel_resp

    async def post_execute(self, retval: Any, code: str, silent: bool) -> None:
        """Post-execution actions

        Handle special kernel variables and display response if not silent.
        """
        # Handle in's
        self.set_variable("_iii", self._iii)
        self.set_variable("_ii", self._ii)
        self.set_variable("_i", code)
        self.set_variable("_i" + str(self.execution_count), code)
        self._iii = self._ii
        self._ii = code
        if retval is not None:
            # --------------------------------------
            # Handle out's (only when non-null)
            self.set_variable("___", self.___)
            self.set_variable("__", self.__)
            self.set_variable("_", retval)
            self.set_variable("_" + str(self.execution_count), retval)
            self.___ = self.__
            self.__ = retval
            self.log.debug(retval)
            if isinstance(retval, ExceptionWrapper):
                self.kernel_resp["status"] = "error"
                content = {
                    "traceback": retval.traceback,
                    "evalue": retval.evalue,
                    "ename": retval.ename,
                }
                self.kernel_resp.update(content)
                if not silent:
                    self.send_response(self.iopub_socket, "error", content)
            elif _is_mime_bundle(retval):
                if not silent:
                    content = {
                        "execution_count": self.execution_count,
                        "data": retval,
                        "metadata": {},
                    }
                    self.send_response(self.iopub_socket, "execute_result", content)
            else:
                try:
                    data = self._display_formatter.format(retval)  # type:ignore[no-untyped-call]
                except Exception as e:
                    self.Error(e)
                    return
                content = {
                    "execution_count": self.execution_count,
                    "data": data[0],
                    "metadata": data[1],
                }
                if not silent:
                    if Widget and isinstance(retval, Widget):
                        self.Display(retval)
                        return
                    self.send_response(self.iopub_socket, "execute_result", content)

    async def do_history(
        self,
        hist_access_type: str | None,
        output: str | None,
        raw: bool | None,
        session: Any = None,
        start: int | None = None,
        stop: int | None = None,
        n: int | None = None,
        pattern: Any = None,
        unique: bool = False,
    ) -> dict[str, str | list[Any]]:
        """
        Access history at startup.

        https://jupyter-client.readthedocs.io/en/stable/messaging.html#history
        """
        with open(self.hist_file) as fid:
            self.hist_cache = json.loads(fid.read() or "[]")
        return {"status": "ok", "history": [(None, None, h) for h in self.hist_cache]}

    async def do_shutdown(self, restart: bool) -> dict[str, Any]:
        """
        Shut down the app gracefully, saving history.

        https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-shutdown
        """
        if self.hist_file:
            with open(self.hist_file, "w") as fid:
                json.dump(self.hist_cache[-self.max_hist_cache :], fid)
        if restart:
            self.Print("Restarting kernel...")
            self.restart_kernel()
            self.reload_magics()
            self.Print("Done!")
        return {"status": "ok", "restart": restart}

    async def do_is_complete(self, code: str) -> dict[str, str]:
        """
        Given code as string, returns dictionary with 'status' representing
        whether code is ready to evaluate. Possible values for status are:

           'complete'   - ready to evaluate
           'incomplete' - not yet ready
           'invalid'    - invalid code
           'unknown'    - unknown; the default unless overridden

        Optionally, if 'status' is 'incomplete', you may indicate
        an indentation string.

        Example:

            return {'status' : 'incomplete',
                    'indent': ' ' * 4}

        https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness
        """
        if code.startswith(self.magic_prefixes["magic"]):
            ## force requirement to end with an empty line
            if code.endswith("\n"):
                return {"status": "complete"}
            else:
                return {"status": "incomplete"}
        # otherwise, how to know is complete?
        elif code.endswith("\n"):
            return {"status": "complete"}
        else:
            return {"status": "incomplete"}

    async def do_complete(self, code: str, cursor_pos: int) -> dict[str, Any]:
        """Handle code completion for the kernel.

        https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion
        """
        info = self.parse_code(code, 0, cursor_pos)
        content = {
            "matches": [],
            "cursor_start": info["start"],
            "cursor_end": info["end"],
            "status": "ok",
            "metadata": {},
        }

        # When the cursor sits after a non-identifier character (e.g. `/`, `-`) the
        # id_regex matches empty string, setting cursor_start to 0.  Path completions
        # are already stripped of their prefix, so we only need to append them at the
        # cursor position — not replace from the beginning of the line.
        if not info["obj"] and info["path_matches"]:
            content["cursor_start"] = content["cursor_end"]

        matches = info["path_matches"]

        if info["magic"]:
            # if the last line contains another magic, use that
            line_info = self.parse_code(info["line"])
            if line_info["magic"]:
                info = line_info

            if info["magic"]["type"] == "line":
                magics = self.line_magics
            else:
                magics = self.cell_magics

            if info["magic"]["name"] in magics:
                magic = magics[info["magic"]["name"]]
                info = info["magic"]
                if info["type"] == "cell" and info["code"]:
                    info = self.parse_code(info["code"])
                else:
                    info = self.parse_code(info["args"])

                matches.extend(magic.get_completions(info))

            elif not info["magic"]["code"] and not info["magic"]["args"]:
                matches = []
                for name in magics.keys():
                    if name.startswith(info["magic"]["name"]):
                        pre = info["magic"]["prefix"]
                        matches.append(pre + name)
                        info["start"] -= len(pre)
                        info["full_obj"] = pre + info["full_obj"]
                        info["obj"] = pre + info["obj"]

        else:
            matches.extend(self.get_completions(info))

        if info["full_obj"] and len(info["full_obj"]) > len(info["obj"]):
            new_list = [m for m in matches if m.startswith(info["full_obj"])]
            if new_list:
                content["cursor_end"] = (
                    content["cursor_end"] + len(info["full_obj"]) - len(info["obj"])
                )
                matches = new_list

        # Extend the path-completion fix (#432) to magic get_completions: when the
        # cursor sits after a non-identifier character (e.g. `--`, `-`) the id_regex
        # again matches empty string so cursor_start is 0.  Completions should be
        # appended at the cursor, not used to replace the entire line.
        if not info["obj"] and matches:
            content["cursor_start"] = content["cursor_end"]

        content["matches"] = sorted(matches)

        return content

    async def do_inspect(
        self, code: str, cursor_pos: int, detail_level: int = 0, omit_sections: Any = ()
    ) -> dict[str, Any] | None:
        """Object introspection.

        https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection
        """
        if cursor_pos > len(code):
            return None

        content = {"status": "aborted", "data": {}, "found": False, "metadata": {}}
        docstring = self.get_help_on(
            code, detail_level, none_on_fail=True, cursor_pos=cursor_pos
        )

        if docstring:
            content["status"] = "ok"
            content["found"] = True
            if isinstance(docstring, dict):  ## {"text/plain": ..., mime-type: ...}
                content["data"] = docstring
                self.log.debug(str(docstring))
            else:
                content["data"] = {"text/plain": docstring}
                self.log.debug(docstring)

        return content

    async def clear_output(self, wait: bool = False) -> None:
        """Clear the output of the kernel."""
        self.send_response(self.iopub_socket, "clear_output", {"wait": wait})

    def DisplayData(
        self, data: dict[str, Any], metadata: dict[str, Any] | None = None
    ) -> None:
        """Display a raw MIME bundle directly without going through IPython's formatter.

        Use this when your kernel produces display data in MIME format directly,
        rather than Python objects with ``_repr_*_`` methods. This is the recommended
        approach for non-Python kernels (e.g. C++, Julia) that generate rich output.

        Args:
            data: A dict mapping MIME types to content.
                  Example: ``{'text/html': '<b>hello</b>', 'text/plain': 'hello'}``
            metadata: Optional dict of per-MIME-type metadata.

        Example::

            kernel.DisplayData(
                {'text/html': '<b>Rich output</b>', 'text/plain': 'Rich output'},
                metadata={'text/html': {'isolated': True}},
            )
        """
        self.log.debug("DisplayData: %s", list(data.keys()))
        content = {"data": data, "metadata": metadata or {}}
        self.send_response(self.iopub_socket, "display_data", content)

    def Display(self, *objects: Any, **kwargs: Any) -> None:
        """Display one or more objects using rich display.

        Supports a `clear_output` keyword argument that clears the output before displaying.

        If an object is a dict whose keys are all MIME types (strings containing ``/``),
        it is treated as a raw MIME bundle and sent directly — equivalent to calling
        :meth:`DisplayData`.

        See https://ipython.readthedocs.io/en/stable/config/integrating.html?highlight=display#rich-display
        """
        if kwargs.get("clear_output"):
            self.send_response(self.iopub_socket, "clear_output", {"wait": True})

        for item in objects:
            if Widget and isinstance(item, Widget):
                self.log.debug("Display Widget")
                data = {
                    "text/plain": repr(item),
                    "application/vnd.jupyter.widget-view+json": {
                        "version_major": 2,
                        "version_minor": 0,
                        "model_id": item._model_id,
                    },
                }
                content = {"data": data, "metadata": {}}
                self.send_response(self.iopub_socket, "display_data", content)
            elif _is_mime_bundle(item):
                self.log.debug("Display raw MIME bundle")
                self.DisplayData(item)
            else:
                self.log.debug("Display Data")
                try:
                    data = self._display_formatter.format(item)  # type:ignore[no-untyped-call]
                except Exception as e:
                    self.Error(e)
                    return
                content = {"data": data[0], "metadata": data[1]}
                self.send_response(self.iopub_socket, "display_data", content)

    def Print(self, *objects: Any, **kwargs: Any) -> None:
        """Print `objects` to the iopub stream, separated by `sep` and followed by `end`.

        Items can be strings or `Widget` instances.
        """
        for item in objects:
            if Widget and isinstance(item, Widget):
                self.Display(item)

        non_widgets = [i for i in objects if not (Widget and isinstance(i, Widget))]
        message = format_message(*non_widgets, **kwargs)

        stream_content = {"name": "stdout", "text": message}
        self.log.debug(f"Print: {message.rstrip()}")
        if self.redirect_to_log:
            self.log.info(message.rstrip())
        else:
            self.send_response(self.iopub_socket, "stream", stream_content)

    def Write(self, message: str) -> None:
        """Write message directly to the iopub stdout with no added end character."""
        stream_content = {"name": "stdout", "text": message}
        self.log.debug(f"Write: {message}")
        if self.redirect_to_log:
            self.log.info(message)
        else:
            self.send_response(self.iopub_socket, "stream", stream_content)

    def Error(self, *objects: Any, **kwargs: Any) -> None:
        """Print `objects` to stdout, separated by `sep` and followed by `end`.

        Objects are cast to strings.
        """
        message = format_message(*objects, **kwargs)
        self.log.debug(f"Error: {message.rstrip()}")
        stream_content = {"name": "stderr", "text": RED + message + NORMAL}
        if self.redirect_to_log:
            self.log.info(message.rstrip())
        else:
            self.send_response(self.iopub_socket, "stream", stream_content)

    def Error_display(self, *objects: Any, **kwargs: Any) -> None:
        """Print `objects` to stdout is they area strings, separated by `sep` and followed by `end`.
        All other objects are rendered using the Display method
        Objects are cast to strings.
        """
        msg = []
        msg_dict = {}
        for item in objects:
            if not isinstance(item, str):
                self.log.debug(f"Item type:{type(item)}")
                self.Display(item)
            else:
                # msg is the error for str
                msg.append(item)

        for k, v in kwargs.items():
            if not isinstance(v, str):
                self.Display(k, v)
            else:
                msg_dict[k] = v

        message = format_message(" ".join(msg), **kwargs)
        if len(msg_dict.keys()) > 0:
            message = format_message(" ".join(msg), msg_dict)
        self.log.debug(f"Error: {message.rstrip()}")
        stream_content = {"name": "stderr", "text": RED + message + NORMAL}
        if self.redirect_to_log:
            self.log.info(message.rstrip())
        else:
            self.send_response(self.iopub_socket, "stream", stream_content)

    def schedule_display_output(self, callback: Callable[[], None]) -> None:
        """Schedule a display output callback to run on the kernel's main IO loop.

        Use this to send output to the frontend from a background thread when
        no execution is in progress. The callback is executed on the kernel's
        main IO loop, ensuring thread safety with ZMQ sockets.

        Example::

            import threading
            import time

            def background_task(kernel):
                while True:
                    time.sleep(10)
                    kernel.schedule_display_output(
                        lambda: kernel.Print("Periodic update from background!")
                    )

            threading.Thread(target=background_task, args=(self,), daemon=True).start()

        Args:
            callback: A callable that will be invoked on the main IO loop.
                      Typically calls methods like :meth:`Print`, :meth:`Write`,
                      :meth:`Display`, :meth:`DisplayData`, or :meth:`Error`.
        """
        io_loop = getattr(self, "io_loop", None)
        if io_loop is not None:
            io_loop.add_callback(callback)
        else:
            callback()

    ##############################
    # Private API and methods not likely to be overridden

    def reload_magics(self) -> None:
        """Reload all of the line and cell magics."""
        self.line_magics: dict[str, Any] = {}
        self.cell_magics: dict[str, Any] = {}
        self.magic_load_errors = []

        # get base magic files and those relative to the current class
        # directory
        magic_files = []
        # Make a metakernel/magics if it doesn't exist:
        local_magics_dir = get_local_magics_dir()
        # Search all of the places there could be magics:
        try:
            paths = [
                os.path.join(
                    os.path.dirname(os.path.abspath(inspect.getfile(self.__class__))),
                    "magics",
                )
            ]
        except Exception:
            paths = []
        paths += [
            local_magics_dir,
            os.path.join(os.path.dirname(os.path.abspath(__file__)), "magics"),
        ]
        self.magic_search_paths = list(paths)
        for magic_dir in paths:
            sys.path.append(magic_dir)
            magic_files.extend(glob.glob(os.path.join(magic_dir, "*.py")))

        for magic in magic_files:
            basename = os.path.basename(magic)
            if basename == "__init__.py":
                continue
            try:
                module = __import__(os.path.splitext(basename)[0])
                importlib.reload(module)
                module.register_magics(self)
            except Exception as e:
                self.log.error(f"Can't load '{magic}': error: {e}")
                self.magic_load_errors.append((magic, str(e)))

    def register_magics(self, magic_klass: type[Magic]) -> None:
        """Register magics for a given magic_klass."""
        magic = magic_klass(self)
        line_magics = magic.get_magics("line")
        cell_magics = magic.get_magics("cell")
        for name in line_magics:
            self.line_magics[name] = magic
        for name in cell_magics:
            self.cell_magics[name] = magic

    def send_response(self, *args: Any, **kwargs: Any) -> None:
        ### if we are running via %parallel, we might not have a
        ### session
        if self.session:
            super().send_response(*args, **kwargs)  # type:ignore[no-untyped-call]

    def call_magic(self, line: str) -> Magic:
        """
        Given an line, such as "%download http://example.com/", parse
        and execute magic.
        """
        return self.get_magic(line)

    def get_magic(self, text: str) -> Magic:
        ## FIXME: Bad name, use call_magic instead.
        # if first line matches a magic,
        # call magic.call_magic() and return magic object
        info = self.parse_code(text)
        magic = self.line_magics["magic"]
        return magic.get_magic(info)  # type:ignore[no-any-return]

    def get_magic_args(self, text: str) -> Magic:
        # if first line matches a magic,
        # call magic.call_magic() and return magic args
        info = self.parse_code(text)
        magic = self.line_magics["magic"]
        return magic.get_magic(info, get_args=True)  # type:ignore[no-any-return]

    def get_help_on(
        self,
        expr: str,
        level: int = 0,
        none_on_fail: bool = False,
        cursor_pos: int = -1,
    ) -> Any:
        """Get help for an expression using the help magic."""
        help_magic = self.line_magics["help"]
        return help_magic.get_text_help_on(expr, level, none_on_fail, cursor_pos)

    def parse_code(
        self, code: str, cursor_start: int = 0, cursor_end: int = -1
    ) -> dict[str, Any]:
        """Parse code using our parser."""
        return self.parser.parse_code(code, cursor_start, cursor_end)

    def _get_sticky_magics(self) -> str:
        retval = ""
        for key in self.sticky_magics:
            retval += key + " " + self.sticky_magics[key] + "\n"
        return retval

    def _send_shell_response(self, socket: Any, stream_type: str, content: Any) -> None:
        publish_display_data({"text/plain": content["text"]})  # type:ignore[no-untyped-call]

Functions

run_as_main(*args: Any, **kwargs: Any) -> None classmethod

Launch or install a metakernel.

Modules implementing a metakernel subclass can use the following lines:

if __name__ == '__main__':
    MetaKernelSubclass.run_as_main()

Source code in metakernel/_metakernel.py

@classmethod
def run_as_main(cls, *args: Any, **kwargs: Any) -> None:
    """Launch or install a metakernel.

    Modules implementing a metakernel subclass can use the following lines:

        if __name__ == '__main__':
            MetaKernelSubclass.run_as_main()
    """
    kwargs["app_name"] = cls.app_name
    MetaKernelApp.launch_instance(kernel_class=cls, *args, **kwargs)  # noqa: B026

makeSubkernel(kernel: MetaKernel) -> None

Run this method in an IPython kernel to set this kernel's input/output settings.

Source code in metakernel/_metakernel.py

def makeSubkernel(self, kernel: MetaKernel) -> None:
    """
    Run this method in an IPython kernel to set
    this kernel's input/output settings.
    """
    from IPython.display import display

    shell = get_ipython()
    if shell:  # we are running under an IPython kernel
        self.session = shell.kernel.session  # type: ignore[attr-defined]
        self.Display = display  # type: ignore[method-assign]
        self.send_response = self._send_shell_response  # type: ignore[method-assign]
    else:
        self.session = kernel.session
        self.send_response = kernel.send_response  # type: ignore[method-assign]
        self.Display = kernel.Display  # type: ignore[method-assign]

set_variable(name: str, value: Any) -> None

Set a variable to a Python-typed value.

Source code in metakernel/_metakernel.py

def set_variable(self, name: str, value: Any) -> None:
    """
    Set a variable to a Python-typed value.
    """
    pass

get_variable(name: str) -> Any

Lookup a variable name and return a Python-typed value.

Source code in metakernel/_metakernel.py

def get_variable(self, name: str) -> Any:
    """
    Lookup a variable name and return a Python-typed value.
    """
    pass

repr(item: Any) -> str

The repr of the kernel.

Source code in metakernel/_metakernel.py

def repr(self, item: Any) -> str:
    """The repr of the kernel."""
    return repr(item)

get_usage() -> str

Get the usage statement for the kernel.

Source code in metakernel/_metakernel.py

def get_usage(self) -> str:
    """Get the usage statement for the kernel."""
    return "This is a usage statement."

get_kernel_help_on(info: dict[str, Any], level: int = 0, none_on_fail: bool = False) -> str | None

Get help on an object. Called by the help magic.

Source code in metakernel/_metakernel.py

def get_kernel_help_on(
    self, info: dict[str, Any], level: int = 0, none_on_fail: bool = False
) -> str | None:
    """Get help on an object.  Called by the help magic."""
    if none_on_fail:
        return None
    else:
        return "Sorry, no help is available on '{}'.".format(info["code"])

handle_plot_settings() -> None

Handle the current plot settings

Source code in metakernel/_metakernel.py

def handle_plot_settings(self) -> None:
    """Handle the current plot settings"""
    pass

get_local_magics_dir() -> str

Returns the path to local magics dir (eg ~/.ipython/metakernel/magics)

Source code in metakernel/_metakernel.py

def get_local_magics_dir(self) -> str:
    """
    Returns the path to local magics dir (eg ~/.ipython/metakernel/magics)
    """
    base = get_ipython_dir()
    return os.path.join(base, "metakernel", "magics")

get_completions(info: dict[str, Any]) -> list[str]

Get completions from kernel based on info dict.

Source code in metakernel/_metakernel.py

def get_completions(self, info: dict[str, Any]) -> list[str]:
    """
    Get completions from kernel based on info dict.
    """
    return []

do_execute_direct(code: str, silent: bool = False) -> Any

Execute code in the kernel language.

Source code in metakernel/_metakernel.py

def do_execute_direct(self, code: str, silent: bool = False) -> Any:
    """
    Execute code in the kernel language.
    """
    pass

do_execute_file(filename: str) -> Any

Default code for running a file. Just opens the file, and sends the text to do_execute_direct.

Source code in metakernel/_metakernel.py

def do_execute_file(self, filename: str) -> Any:
    """
    Default code for running a file. Just opens the file, and sends
    the text to do_execute_direct.
    """
    with open(filename) as f:
        return self.do_execute_direct("".join(f.readlines()))

do_execute_meta(code: str) -> Any

Execute meta code in the kernel. This uses the execute infrastructure but allows JavaScript to talk directly to the kernel bypassing normal processing.

When responding to the %%debug magic, the step and reset meta commands can answer with a string in the format:

"highlight: [start_line, start_col, end_line, end_col]"

for highlighting expressions in the frontend.

Source code in metakernel/_metakernel.py

def do_execute_meta(self, code: str) -> Any:
    """
    Execute meta code in the kernel. This uses the execute infrastructure
    but allows JavaScript to talk directly to the kernel bypassing normal
    processing.

    When responding to the %%debug magic, the step and reset meta
    commands can answer with a string in the format:

    "highlight: [start_line, start_col, end_line, end_col]"

    for highlighting expressions in the frontend.
    """
    if code == "reset":
        raise Exception("This kernel does not implement this meta command")
    elif code == "stop":
        raise Exception("This kernel does not implement this meta command")
    elif code == "step":
        raise Exception("This kernel does not implement this meta command")
    elif code.startswith("inspect "):
        raise Exception("This kernel does not implement this meta command")
    else:
        raise Exception(f"Unknown meta command: '{code}'")

initialize_debug(code: str) -> str

This function is used with the %%debug magic for highlighting lines of code, and for initializing debug functions.

Return the empty string if highlighting is not supported.

Source code in metakernel/_metakernel.py

def initialize_debug(self, code: str) -> str:
    """
    This function is used with the %%debug magic for highlighting
    lines of code, and for initializing debug functions.

    Return the empty string if highlighting is not supported.
    """
    # return "highlight: [%s, %s, %s, %s]" % (line1, col1, line2, col2)
    return ""

do_function_direct(function_name: str, arg: Any) -> Any

Call a function in the kernel language with args (as a single item).

Source code in metakernel/_metakernel.py

def do_function_direct(self, function_name: str, arg: Any) -> Any:
    """
    Call a function in the kernel language with args (as a single item).
    """
    f = self.do_execute_direct(function_name)
    return f(arg)

restart_kernel() -> None

Restart the kernel

Source code in metakernel/_metakernel.py

def restart_kernel(self) -> None:
    """Restart the kernel"""
    pass

do_execute(code: Any, silent: Any = False, store_history: Any = True, user_expressions: Any = None, allow_stdin: Any = False, *, cell_meta: Any = None, cell_id: Any = None) -> Any async

Handle code execution.

https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute

Source code in metakernel/_metakernel.py

async def do_execute(
    self,
    code: Any,
    silent: Any = False,
    store_history: Any = True,
    user_expressions: Any = None,
    allow_stdin: Any = False,
    *,
    cell_meta: Any = None,
    cell_id: Any = None,
) -> Any:
    """Handle code execution.

    https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute
    """
    # Set the ability for the kernel to get standard-in:
    self._allow_stdin = allow_stdin
    # Create a default response:
    self.kernel_resp = {
        "status": "ok",
        # The base class increments the execution count
        "execution_count": self.execution_count,
        "payload": [],
        "user_expressions": {},
    }

    # TODO: remove this when IPython fixes this
    # This happens at startup when the language is set to python
    if "_usage.page_guiref" in code:
        return self.kernel_resp

    if code and store_history:
        self.hist_cache.append(code.strip())

    if not code.strip():
        return self.kernel_resp

    info = self.parse_code(code)
    self.payload = []
    retval = None

    if info["magic"] and info["magic"]["name"] == "help":
        if info["magic"]["type"] == "line":
            level = 0
        else:
            level = 1
        text = self.get_help_on(code, level)
        if text:
            content = {
                "start_line_number": 0,
                "source": "page",
            }
            if isinstance(text, dict):
                content["data"] = text  ## {mime-type: ..., mime-type:...}
                self.log.debug(str(text))
            else:
                content["data"] = {"text/plain": text}
                self.log.debug(text)
            self.payload = [content]

    elif info["magic"] or self.sticky_magics:
        retval = None
        if self.sticky_magics:
            magics, code = _split_magics_code(code, self.magic_prefixes)
            code = magics + self._get_sticky_magics() + code
        stack = []
        # Handle magics:
        magic = None
        prefixes = (self.magic_prefixes["shell"], self.magic_prefixes["magic"])
        while code.startswith(prefixes):
            magic = self.get_magic(code)
            if magic is not None:
                stack.append(magic)
                code = str(magic.get_code())
                # signal to exit, maybe error or no block
                if not magic.evaluate:
                    break
            else:
                break  # type:ignore[unreachable]
        # Execute code, if any:
        if (magic is None or magic.evaluate) and code.strip() != "":
            if code.startswith("~~META~~:"):
                retval = self.do_execute_meta(code[9:].strip())
            else:
                try:
                    retval = self.do_execute_direct(code)
                    if inspect.isawaitable(retval):
                        retval = await retval
                except SystemExit:
                    self._request_shutdown()
                    return self.kernel_resp
                except Exception as e:
                    retval = ExceptionWrapper(type(e).__name__, str(e), [])
        # Post-process magics:
        for magic in reversed(stack):
            retval = magic.post_process(retval)
    else:
        if code.startswith("~~META~~:"):
            retval = self.do_execute_meta(code[9:].strip())
        else:
            try:
                retval = self.do_execute_direct(code)
                if inspect.isawaitable(retval):
                    retval = await retval
            except SystemExit:
                self._request_shutdown()
                return self.kernel_resp
            except Exception as e:
                retval = ExceptionWrapper(type(e).__name__, str(e), [])

    await self.post_execute(retval, code, silent)

    if "payload" in self.kernel_resp:
        self.kernel_resp["payload"] = self.payload

    return self.kernel_resp

post_execute(retval: Any, code: str, silent: bool) -> None async

Post-execution actions

Handle special kernel variables and display response if not silent.

Source code in metakernel/_metakernel.py

async def post_execute(self, retval: Any, code: str, silent: bool) -> None:
    """Post-execution actions

    Handle special kernel variables and display response if not silent.
    """
    # Handle in's
    self.set_variable("_iii", self._iii)
    self.set_variable("_ii", self._ii)
    self.set_variable("_i", code)
    self.set_variable("_i" + str(self.execution_count), code)
    self._iii = self._ii
    self._ii = code
    if retval is not None:
        # --------------------------------------
        # Handle out's (only when non-null)
        self.set_variable("___", self.___)
        self.set_variable("__", self.__)
        self.set_variable("_", retval)
        self.set_variable("_" + str(self.execution_count), retval)
        self.___ = self.__
        self.__ = retval
        self.log.debug(retval)
        if isinstance(retval, ExceptionWrapper):
            self.kernel_resp["status"] = "error"
            content = {
                "traceback": retval.traceback,
                "evalue": retval.evalue,
                "ename": retval.ename,
            }
            self.kernel_resp.update(content)
            if not silent:
                self.send_response(self.iopub_socket, "error", content)
        elif _is_mime_bundle(retval):
            if not silent:
                content = {
                    "execution_count": self.execution_count,
                    "data": retval,
                    "metadata": {},
                }
                self.send_response(self.iopub_socket, "execute_result", content)
        else:
            try:
                data = self._display_formatter.format(retval)  # type:ignore[no-untyped-call]
            except Exception as e:
                self.Error(e)
                return
            content = {
                "execution_count": self.execution_count,
                "data": data[0],
                "metadata": data[1],
            }
            if not silent:
                if Widget and isinstance(retval, Widget):
                    self.Display(retval)
                    return
                self.send_response(self.iopub_socket, "execute_result", content)

do_history(hist_access_type: str | None, output: str | None, raw: bool | None, session: Any = None, start: int | None = None, stop: int | None = None, n: int | None = None, pattern: Any = None, unique: bool = False) -> dict[str, str | list[Any]] async

Access history at startup.

https://jupyter-client.readthedocs.io/en/stable/messaging.html#history

Source code in metakernel/_metakernel.py

async def do_history(
    self,
    hist_access_type: str | None,
    output: str | None,
    raw: bool | None,
    session: Any = None,
    start: int | None = None,
    stop: int | None = None,
    n: int | None = None,
    pattern: Any = None,
    unique: bool = False,
) -> dict[str, str | list[Any]]:
    """
    Access history at startup.

    https://jupyter-client.readthedocs.io/en/stable/messaging.html#history
    """
    with open(self.hist_file) as fid:
        self.hist_cache = json.loads(fid.read() or "[]")
    return {"status": "ok", "history": [(None, None, h) for h in self.hist_cache]}

do_shutdown(restart: bool) -> dict[str, Any] async

Shut down the app gracefully, saving history.

https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-shutdown

Source code in metakernel/_metakernel.py

async def do_shutdown(self, restart: bool) -> dict[str, Any]:
    """
    Shut down the app gracefully, saving history.

    https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-shutdown
    """
    if self.hist_file:
        with open(self.hist_file, "w") as fid:
            json.dump(self.hist_cache[-self.max_hist_cache :], fid)
    if restart:
        self.Print("Restarting kernel...")
        self.restart_kernel()
        self.reload_magics()
        self.Print("Done!")
    return {"status": "ok", "restart": restart}

do_is_complete(code: str) -> dict[str, str] async

Given code as string, returns dictionary with 'status' representing whether code is ready to evaluate. Possible values for status are:

'complete' - ready to evaluate 'incomplete' - not yet ready 'invalid' - invalid code 'unknown' - unknown; the default unless overridden

Optionally, if 'status' is 'incomplete', you may indicate an indentation string.

Example:

return {'status' : 'incomplete',
        'indent': ' ' * 4}

https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness

Source code in metakernel/_metakernel.py

async def do_is_complete(self, code: str) -> dict[str, str]:
    """
    Given code as string, returns dictionary with 'status' representing
    whether code is ready to evaluate. Possible values for status are:

       'complete'   - ready to evaluate
       'incomplete' - not yet ready
       'invalid'    - invalid code
       'unknown'    - unknown; the default unless overridden

    Optionally, if 'status' is 'incomplete', you may indicate
    an indentation string.

    Example:

        return {'status' : 'incomplete',
                'indent': ' ' * 4}

    https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness
    """
    if code.startswith(self.magic_prefixes["magic"]):
        ## force requirement to end with an empty line
        if code.endswith("\n"):
            return {"status": "complete"}
        else:
            return {"status": "incomplete"}
    # otherwise, how to know is complete?
    elif code.endswith("\n"):
        return {"status": "complete"}
    else:
        return {"status": "incomplete"}

do_complete(code: str, cursor_pos: int) -> dict[str, Any] async

Handle code completion for the kernel.

https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion

Source code in metakernel/_metakernel.py

async def do_complete(self, code: str, cursor_pos: int) -> dict[str, Any]:
    """Handle code completion for the kernel.

    https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion
    """
    info = self.parse_code(code, 0, cursor_pos)
    content = {
        "matches": [],
        "cursor_start": info["start"],
        "cursor_end": info["end"],
        "status": "ok",
        "metadata": {},
    }

    # When the cursor sits after a non-identifier character (e.g. `/`, `-`) the
    # id_regex matches empty string, setting cursor_start to 0.  Path completions
    # are already stripped of their prefix, so we only need to append them at the
    # cursor position — not replace from the beginning of the line.
    if not info["obj"] and info["path_matches"]:
        content["cursor_start"] = content["cursor_end"]

    matches = info["path_matches"]

    if info["magic"]:
        # if the last line contains another magic, use that
        line_info = self.parse_code(info["line"])
        if line_info["magic"]:
            info = line_info

        if info["magic"]["type"] == "line":
            magics = self.line_magics
        else:
            magics = self.cell_magics

        if info["magic"]["name"] in magics:
            magic = magics[info["magic"]["name"]]
            info = info["magic"]
            if info["type"] == "cell" and info["code"]:
                info = self.parse_code(info["code"])
            else:
                info = self.parse_code(info["args"])

            matches.extend(magic.get_completions(info))

        elif not info["magic"]["code"] and not info["magic"]["args"]:
            matches = []
            for name in magics.keys():
                if name.startswith(info["magic"]["name"]):
                    pre = info["magic"]["prefix"]
                    matches.append(pre + name)
                    info["start"] -= len(pre)
                    info["full_obj"] = pre + info["full_obj"]
                    info["obj"] = pre + info["obj"]

    else:
        matches.extend(self.get_completions(info))

    if info["full_obj"] and len(info["full_obj"]) > len(info["obj"]):
        new_list = [m for m in matches if m.startswith(info["full_obj"])]
        if new_list:
            content["cursor_end"] = (
                content["cursor_end"] + len(info["full_obj"]) - len(info["obj"])
            )
            matches = new_list

    # Extend the path-completion fix (#432) to magic get_completions: when the
    # cursor sits after a non-identifier character (e.g. `--`, `-`) the id_regex
    # again matches empty string so cursor_start is 0.  Completions should be
    # appended at the cursor, not used to replace the entire line.
    if not info["obj"] and matches:
        content["cursor_start"] = content["cursor_end"]

    content["matches"] = sorted(matches)

    return content

do_inspect(code: str, cursor_pos: int, detail_level: int = 0, omit_sections: Any = ()) -> dict[str, Any] | None async

Object introspection.

https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection

Source code in metakernel/_metakernel.py

async def do_inspect(
    self, code: str, cursor_pos: int, detail_level: int = 0, omit_sections: Any = ()
) -> dict[str, Any] | None:
    """Object introspection.

    https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection
    """
    if cursor_pos > len(code):
        return None

    content = {"status": "aborted", "data": {}, "found": False, "metadata": {}}
    docstring = self.get_help_on(
        code, detail_level, none_on_fail=True, cursor_pos=cursor_pos
    )

    if docstring:
        content["status"] = "ok"
        content["found"] = True
        if isinstance(docstring, dict):  ## {"text/plain": ..., mime-type: ...}
            content["data"] = docstring
            self.log.debug(str(docstring))
        else:
            content["data"] = {"text/plain": docstring}
            self.log.debug(docstring)

    return content

clear_output(wait: bool = False) -> None async

Clear the output of the kernel.

Source code in metakernel/_metakernel.py

async def clear_output(self, wait: bool = False) -> None:
    """Clear the output of the kernel."""
    self.send_response(self.iopub_socket, "clear_output", {"wait": wait})

DisplayData(data: dict[str, Any], metadata: dict[str, Any] | None = None) -> None

Display a raw MIME bundle directly without going through IPython's formatter.

Use this when your kernel produces display data in MIME format directly, rather than Python objects with _repr_*_ methods. This is the recommended approach for non-Python kernels (e.g. C++, Julia) that generate rich output.

Args: data: A dict mapping MIME types to content. Example: {'text/html': '<b>hello</b>', 'text/plain': 'hello'} metadata: Optional dict of per-MIME-type metadata.

Example::

kernel.DisplayData(
    {'text/html': '<b>Rich output</b>', 'text/plain': 'Rich output'},
    metadata={'text/html': {'isolated': True}},
)

Source code in metakernel/_metakernel.py

def DisplayData(
    self, data: dict[str, Any], metadata: dict[str, Any] | None = None
) -> None:
    """Display a raw MIME bundle directly without going through IPython's formatter.

    Use this when your kernel produces display data in MIME format directly,
    rather than Python objects with ``_repr_*_`` methods. This is the recommended
    approach for non-Python kernels (e.g. C++, Julia) that generate rich output.

    Args:
        data: A dict mapping MIME types to content.
              Example: ``{'text/html': '<b>hello</b>', 'text/plain': 'hello'}``
        metadata: Optional dict of per-MIME-type metadata.

    Example::

        kernel.DisplayData(
            {'text/html': '<b>Rich output</b>', 'text/plain': 'Rich output'},
            metadata={'text/html': {'isolated': True}},
        )
    """
    self.log.debug("DisplayData: %s", list(data.keys()))
    content = {"data": data, "metadata": metadata or {}}
    self.send_response(self.iopub_socket, "display_data", content)

Display(*objects: Any, **kwargs: Any) -> None

Display one or more objects using rich display.

Supports a clear_output keyword argument that clears the output before displaying.

If an object is a dict whose keys are all MIME types (strings containing /), it is treated as a raw MIME bundle and sent directly — equivalent to calling :meth:DisplayData.

See https://ipython.readthedocs.io/en/stable/config/integrating.html?highlight=display#rich-display

Source code in metakernel/_metakernel.py

def Display(self, *objects: Any, **kwargs: Any) -> None:
    """Display one or more objects using rich display.

    Supports a `clear_output` keyword argument that clears the output before displaying.

    If an object is a dict whose keys are all MIME types (strings containing ``/``),
    it is treated as a raw MIME bundle and sent directly — equivalent to calling
    :meth:`DisplayData`.

    See https://ipython.readthedocs.io/en/stable/config/integrating.html?highlight=display#rich-display
    """
    if kwargs.get("clear_output"):
        self.send_response(self.iopub_socket, "clear_output", {"wait": True})

    for item in objects:
        if Widget and isinstance(item, Widget):
            self.log.debug("Display Widget")
            data = {
                "text/plain": repr(item),
                "application/vnd.jupyter.widget-view+json": {
                    "version_major": 2,
                    "version_minor": 0,
                    "model_id": item._model_id,
                },
            }
            content = {"data": data, "metadata": {}}
            self.send_response(self.iopub_socket, "display_data", content)
        elif _is_mime_bundle(item):
            self.log.debug("Display raw MIME bundle")
            self.DisplayData(item)
        else:
            self.log.debug("Display Data")
            try:
                data = self._display_formatter.format(item)  # type:ignore[no-untyped-call]
            except Exception as e:
                self.Error(e)
                return
            content = {"data": data[0], "metadata": data[1]}
            self.send_response(self.iopub_socket, "display_data", content)

Print(*objects: Any, **kwargs: Any) -> None

Print objects to the iopub stream, separated by sep and followed by end.

Items can be strings or Widget instances.

Source code in metakernel/_metakernel.py

def Print(self, *objects: Any, **kwargs: Any) -> None:
    """Print `objects` to the iopub stream, separated by `sep` and followed by `end`.

    Items can be strings or `Widget` instances.
    """
    for item in objects:
        if Widget and isinstance(item, Widget):
            self.Display(item)

    non_widgets = [i for i in objects if not (Widget and isinstance(i, Widget))]
    message = format_message(*non_widgets, **kwargs)

    stream_content = {"name": "stdout", "text": message}
    self.log.debug(f"Print: {message.rstrip()}")
    if self.redirect_to_log:
        self.log.info(message.rstrip())
    else:
        self.send_response(self.iopub_socket, "stream", stream_content)

Write(message: str) -> None

Write message directly to the iopub stdout with no added end character.

Source code in metakernel/_metakernel.py

def Write(self, message: str) -> None:
    """Write message directly to the iopub stdout with no added end character."""
    stream_content = {"name": "stdout", "text": message}
    self.log.debug(f"Write: {message}")
    if self.redirect_to_log:
        self.log.info(message)
    else:
        self.send_response(self.iopub_socket, "stream", stream_content)

Error(*objects: Any, **kwargs: Any) -> None

Print objects to stdout, separated by sep and followed by end.

Objects are cast to strings.

Source code in metakernel/_metakernel.py

def Error(self, *objects: Any, **kwargs: Any) -> None:
    """Print `objects` to stdout, separated by `sep` and followed by `end`.

    Objects are cast to strings.
    """
    message = format_message(*objects, **kwargs)
    self.log.debug(f"Error: {message.rstrip()}")
    stream_content = {"name": "stderr", "text": RED + message + NORMAL}
    if self.redirect_to_log:
        self.log.info(message.rstrip())
    else:
        self.send_response(self.iopub_socket, "stream", stream_content)

Error_display(*objects: Any, **kwargs: Any) -> None

Print objects to stdout is they area strings, separated by sep and followed by end. All other objects are rendered using the Display method Objects are cast to strings.

Source code in metakernel/_metakernel.py

def Error_display(self, *objects: Any, **kwargs: Any) -> None:
    """Print `objects` to stdout is they area strings, separated by `sep` and followed by `end`.
    All other objects are rendered using the Display method
    Objects are cast to strings.
    """
    msg = []
    msg_dict = {}
    for item in objects:
        if not isinstance(item, str):
            self.log.debug(f"Item type:{type(item)}")
            self.Display(item)
        else:
            # msg is the error for str
            msg.append(item)

    for k, v in kwargs.items():
        if not isinstance(v, str):
            self.Display(k, v)
        else:
            msg_dict[k] = v

    message = format_message(" ".join(msg), **kwargs)
    if len(msg_dict.keys()) > 0:
        message = format_message(" ".join(msg), msg_dict)
    self.log.debug(f"Error: {message.rstrip()}")
    stream_content = {"name": "stderr", "text": RED + message + NORMAL}
    if self.redirect_to_log:
        self.log.info(message.rstrip())
    else:
        self.send_response(self.iopub_socket, "stream", stream_content)

schedule_display_output(callback: Callable[[], None]) -> None

Schedule a display output callback to run on the kernel's main IO loop.

Use this to send output to the frontend from a background thread when no execution is in progress. The callback is executed on the kernel's main IO loop, ensuring thread safety with ZMQ sockets.

Example::

import threading
import time

def background_task(kernel):
    while True:
        time.sleep(10)
        kernel.schedule_display_output(
            lambda: kernel.Print("Periodic update from background!")
        )

threading.Thread(target=background_task, args=(self,), daemon=True).start()

Args: callback: A callable that will be invoked on the main IO loop. Typically calls methods like :meth:Print, :meth:Write, :meth:Display, :meth:DisplayData, or :meth:Error.

Source code in metakernel/_metakernel.py

def schedule_display_output(self, callback: Callable[[], None]) -> None:
    """Schedule a display output callback to run on the kernel's main IO loop.

    Use this to send output to the frontend from a background thread when
    no execution is in progress. The callback is executed on the kernel's
    main IO loop, ensuring thread safety with ZMQ sockets.

    Example::

        import threading
        import time

        def background_task(kernel):
            while True:
                time.sleep(10)
                kernel.schedule_display_output(
                    lambda: kernel.Print("Periodic update from background!")
                )

        threading.Thread(target=background_task, args=(self,), daemon=True).start()

    Args:
        callback: A callable that will be invoked on the main IO loop.
                  Typically calls methods like :meth:`Print`, :meth:`Write`,
                  :meth:`Display`, :meth:`DisplayData`, or :meth:`Error`.
    """
    io_loop = getattr(self, "io_loop", None)
    if io_loop is not None:
        io_loop.add_callback(callback)
    else:
        callback()

reload_magics() -> None

Reload all of the line and cell magics.

Source code in metakernel/_metakernel.py

def reload_magics(self) -> None:
    """Reload all of the line and cell magics."""
    self.line_magics: dict[str, Any] = {}
    self.cell_magics: dict[str, Any] = {}
    self.magic_load_errors = []

    # get base magic files and those relative to the current class
    # directory
    magic_files = []
    # Make a metakernel/magics if it doesn't exist:
    local_magics_dir = get_local_magics_dir()
    # Search all of the places there could be magics:
    try:
        paths = [
            os.path.join(
                os.path.dirname(os.path.abspath(inspect.getfile(self.__class__))),
                "magics",
            )
        ]
    except Exception:
        paths = []
    paths += [
        local_magics_dir,
        os.path.join(os.path.dirname(os.path.abspath(__file__)), "magics"),
    ]
    self.magic_search_paths = list(paths)
    for magic_dir in paths:
        sys.path.append(magic_dir)
        magic_files.extend(glob.glob(os.path.join(magic_dir, "*.py")))

    for magic in magic_files:
        basename = os.path.basename(magic)
        if basename == "__init__.py":
            continue
        try:
            module = __import__(os.path.splitext(basename)[0])
            importlib.reload(module)
            module.register_magics(self)
        except Exception as e:
            self.log.error(f"Can't load '{magic}': error: {e}")
            self.magic_load_errors.append((magic, str(e)))

register_magics(magic_klass: type[Magic]) -> None

Register magics for a given magic_klass.

Source code in metakernel/_metakernel.py

def register_magics(self, magic_klass: type[Magic]) -> None:
    """Register magics for a given magic_klass."""
    magic = magic_klass(self)
    line_magics = magic.get_magics("line")
    cell_magics = magic.get_magics("cell")
    for name in line_magics:
        self.line_magics[name] = magic
    for name in cell_magics:
        self.cell_magics[name] = magic

call_magic(line: str) -> Magic

Given an line, such as "%download http://example.com/", parse and execute magic.

Source code in metakernel/_metakernel.py

def call_magic(self, line: str) -> Magic:
    """
    Given an line, such as "%download http://example.com/", parse
    and execute magic.
    """
    return self.get_magic(line)

get_help_on(expr: str, level: int = 0, none_on_fail: bool = False, cursor_pos: int = -1) -> Any

Get help for an expression using the help magic.

Source code in metakernel/_metakernel.py

def get_help_on(
    self,
    expr: str,
    level: int = 0,
    none_on_fail: bool = False,
    cursor_pos: int = -1,
) -> Any:
    """Get help for an expression using the help magic."""
    help_magic = self.line_magics["help"]
    return help_magic.get_text_help_on(expr, level, none_on_fail, cursor_pos)

parse_code(code: str, cursor_start: int = 0, cursor_end: int = -1) -> dict[str, Any]

Parse code using our parser.

Source code in metakernel/_metakernel.py

def parse_code(
    self, code: str, cursor_start: int = 0, cursor_end: int = -1
) -> dict[str, Any]:
    """Parse code using our parser."""
    return self.parser.parse_code(code, cursor_start, cursor_end)

Magic

metakernel.Magic

Base class to define magics for MetaKernel based kernels.

Users can redefine the default magics provided by Metakernel by creating a module with the exact same name as the Metakernel magic.

For example, you can override %matplotlib in your kernel by writing a new magic inside magics/matplotlib_magic.py

Source code in metakernel/magic.py

class Magic:
    """
    Base class to define magics for MetaKernel based kernels.

    Users can redefine the default magics provided by Metakernel
    by creating a module with the exact same name as the
    Metakernel magic.

    For example, you can override %matplotlib in your kernel by
    writing a new magic inside magics/matplotlib_magic.py
    """

    def __init__(self, kernel: MetaKernel) -> None:
        self.kernel = kernel
        self.evaluate = True
        self.code = ""

    def get_args(self, mtype: str, name: str, code: str, args: Any) -> Any:
        self.code = code
        old_args = args
        mtype = mtype.replace("sticky", "cell")

        func = getattr(self, mtype + "_" + name)
        try:
            args, kwargs = _parse_args(func, args, usage=self.get_help(mtype, name))
        except Exception as e:
            self.kernel.Error(str(e))
            return self
        arg_spec = inspect.getfullargspec(func)
        fargs = arg_spec.args
        if fargs[0] == "self":
            fargs = fargs[1:]

        fargs = [f for f in fargs if f not in kwargs.keys()]
        if len(args) > len(fargs) and not arg_spec.varargs:
            extra = " ".join(str(s) for s in (args[len(fargs) - 1 :]))
            args = args[: len(fargs) - 1] + [extra]

        return (args, kwargs, old_args)

    def call_magic(self, mtype: str, name: str, code: str, args: Any) -> Magic:
        self.code = code
        old_args = args
        mtype = mtype.replace("sticky", "cell")

        func = getattr(self, mtype + "_" + name)
        try:
            args, kwargs = _parse_args(func, args, usage=self.get_help(mtype, name))
        except Exception as e:
            self.kernel.Error(str(e))
            return self

        arg_spec = inspect.getfullargspec(func)
        fargs = arg_spec.args
        if fargs[0] == "self":
            fargs = fargs[1:]

        fargs = [f for f in fargs if f not in kwargs.keys()]
        if len(args) > len(fargs) and not arg_spec.varargs:
            extra = " ".join(str(s) for s in (args[len(fargs) - 1 :]))
            args = args[: len(fargs) - 1] + [extra]

        try:
            try:
                func(*args, **kwargs)
            except TypeError:
                func(old_args)
        except Exception as exc:
            msg = f"Error in calling magic '{name}' on {mtype}:\n    {exc!s}\n    args: {args}\n    kwargs: {kwargs}"
            self.kernel.Error(msg)
            self.kernel.Error(traceback.format_exc())
            self.kernel.Error(self.get_help(mtype, name))
            # return dummy magic to end processing:
            return Magic(self.kernel)
        return self

    def get_help(self, mtype: str, name: str, level: int = 0) -> str:
        if hasattr(self, mtype + "_" + name):
            func = getattr(self, mtype + "_" + name)
            if level == 0:
                if func.__doc__:
                    return _trim(func.__doc__)  # type: ignore[return-value]
                else:
                    return f"No help available for magic '{name}' for {mtype}s."
            else:
                filename = inspect.getfile(func)
                if filename and os.path.exists(filename):
                    with open(filename) as f:
                        return f.read()
                else:
                    return f"No help available for magic '{name}' for {mtype}s."
        else:
            return f"No such magic '{name}' for {mtype}s."

    def get_help_on(self, info: dict[str, Any], level: int = 0) -> str | None:
        return "Sorry, no help is available on '{}'.".format(info["code"])

    def get_completions(self, info: dict[str, Any]) -> list[str]:
        """
        Get completions based on info dict from magic.
        """
        return []

    def get_magics(self, mtype: str) -> list[str]:
        magics = []
        for name in dir(self):
            if name.startswith(mtype + "_"):
                magics.append(name.replace(mtype + "_", ""))
        return magics

    def get_code(self) -> str:
        return self.code

    def post_process(self, retval: Any) -> Any:
        return retval

Functions

get_completions(info: dict[str, Any]) -> list[str]

Get completions based on info dict from magic.

Source code in metakernel/magic.py

def get_completions(self, info: dict[str, Any]) -> list[str]:
    """
    Get completions based on info dict from magic.
    """
    return []

option

metakernel.option(*args: Any, **kwargs: Any) -> Callable[[_F], _F]

Return decorator that adds a magic option to a function.

Source code in metakernel/magic.py

def option(*args: Any, **kwargs: Any) -> Callable[[_F], _F]:
    """Return decorator that adds a magic option to a function."""

    def decorator(func: _F) -> _F:
        help_text = ""
        if not getattr(func, "has_options", False):
            func.has_options = True  # type:ignore[attr-defined]
            func.options = []  # type:ignore[attr-defined]
            help_text += "Options:\n-------\n"
        try:
            option = optparse.Option(*args, **kwargs)
        except optparse.OptionError:
            help_text += args[0] + "\n"
        else:
            help_text += _format_option(option) + "\n"
            func.options.append(option)  # type:ignore[attr-defined]
        if func.__doc__:
            func.__doc__ += _indent(func.__doc__, help_text)
        else:
            func.__doc__ = help_text
        return func

    return decorator

ProcessMetaKernel

metakernel.ProcessMetaKernel

Bases: MetaKernel

Source code in metakernel/process_metakernel.py

class ProcessMetaKernel(MetaKernel):
    implementation = "process_kernel"
    implementation_version = __version__
    language = "process"
    language_info: dict[str, Any] = {
        # 'mimetype': 'text/x-python',
        # 'language': 'python',
        # ------ If different from 'language':
        # 'codemirror_mode': 'language',
        # 'pygments_lexer': 'language',
        # 'file_extension': 'py',
    }

    @property
    def language_version(self) -> str | None:
        m = version_pat.search(self.banner)
        if m is None:
            return None
        return m.group(1)

    _banner: str | None = "Process"

    @property
    def banner(self) -> str:  # type:ignore[override]
        assert self._banner is not None  # noqa: S101
        return self._banner

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        MetaKernel.__init__(self, *args, **kwargs)
        self.wrapper: REPLWrapper | None = None

    def do_execute_direct(self, code: str, silent: bool = False) -> TextOutput | None:
        """Execute the code in the subprocess."""
        if not self.wrapper:
            self.wrapper = self.makeWrapper()

        self.payload = []
        wrapper = self.wrapper
        child = wrapper.child

        if not code.strip():
            self.kernel_resp = {
                "status": "ok",
                "execution_count": self.execution_count,
                "payload": [],
                "user_expressions": {},
            }
            return None

        interrupted = False
        output = ""
        error = None
        stream_handler = self.Write if not silent else None
        try:
            output = wrapper.run_command(
                code.rstrip(),
                timeout=None,
                stream_handler=stream_handler,
                stdin_handler=self.raw_input,
            )
        except KeyboardInterrupt:
            interrupted = True
            output = wrapper.interrupt()
        except EOF:
            self.Print(child.before)
            if self.wrapper is not None:
                try:
                    self.wrapper.terminate()
                except Exception:  # noqa: S110
                    pass
            self.restart_kernel()
            self.reload_magics()
            error = RuntimeError("End of File")
            tb = "End of File"
        except Exception as e:
            _ex_type, error, tb = sys.exc_info()  # type: ignore[assignment]
            self.Error(str(e))

        if interrupted:
            self.kernel_resp = {
                "status": "abort",
                "execution_count": self.execution_count,
            }

        exitcode, trace = self.check_exitcode()

        if exitcode:
            self.kernel_resp = {
                "status": "error",
                "execution_count": self.execution_count,
                "ename": "",
                "evalue": str(exitcode),
                "traceback": trace,
            }

        elif error:
            self.kernel_resp = {
                "status": "error",
                "execution_count": self.execution_count,
                "ename": "",
                "evalue": str(error),
                "traceback": str(tb),
            }

        else:
            self.kernel_resp = {
                "status": "ok",
                "execution_count": self.execution_count,
                "payload": [],
                "user_expressions": {},
            }

        if output:
            if stream_handler:
                stream_handler(output)
            else:
                return TextOutput(output)
        return None

    def check_exitcode(self) -> tuple[int, None]:
        """
        Return (1, ["trace"]) if error.
        """
        return (0, None)

    def makeWrapper(self) -> REPLWrapper:
        """
        In this method the REPLWrapper is created and returned.
        REPLWrapper takes the name of the executable, and arguments
        describing the executable prompt:

        return REPLWrapper('bash', orig_prompt, prompt_change,
                           prompt_cmd=prompt_cmd,
                           extra_init_cmd=extra_init_cmd)

        The parameters are:

        :param orig_prompt: What the original prompt is (or is forced
        to be by prompt_cmd).

        :param prompt_cmd: Used when the prompt is not printed by default
        (happens on Windows) to print something that we can search for.

        :param prompt_change: Used to set the PS1/PS2 equivalents to
        something that is easier to tell apart from everything else,
        make sure it is a string with {0} and {1} in it for the
        PS1/PS2 fill-ins.

        See `metakernel.replwrap.REPLWrapper` for more details.

        """
        raise NotImplementedError

    async def do_shutdown(self, restart: bool) -> dict[str, str]:
        """
        Shut down the app gracefully, saving history.
        """
        if self.wrapper is not None:
            try:
                self.wrapper.terminate()
            except Exception as e:
                self.Error(str(e))
        return await super().do_shutdown(restart)

    def restart_kernel(self) -> None:
        """Restart the kernel"""
        self.wrapper = self.makeWrapper()

Functions

do_execute_direct(code: str, silent: bool = False) -> TextOutput | None

Execute the code in the subprocess.

Source code in metakernel/process_metakernel.py

def do_execute_direct(self, code: str, silent: bool = False) -> TextOutput | None:
    """Execute the code in the subprocess."""
    if not self.wrapper:
        self.wrapper = self.makeWrapper()

    self.payload = []
    wrapper = self.wrapper
    child = wrapper.child

    if not code.strip():
        self.kernel_resp = {
            "status": "ok",
            "execution_count": self.execution_count,
            "payload": [],
            "user_expressions": {},
        }
        return None

    interrupted = False
    output = ""
    error = None
    stream_handler = self.Write if not silent else None
    try:
        output = wrapper.run_command(
            code.rstrip(),
            timeout=None,
            stream_handler=stream_handler,
            stdin_handler=self.raw_input,
        )
    except KeyboardInterrupt:
        interrupted = True
        output = wrapper.interrupt()
    except EOF:
        self.Print(child.before)
        if self.wrapper is not None:
            try:
                self.wrapper.terminate()
            except Exception:  # noqa: S110
                pass
        self.restart_kernel()
        self.reload_magics()
        error = RuntimeError("End of File")
        tb = "End of File"
    except Exception as e:
        _ex_type, error, tb = sys.exc_info()  # type: ignore[assignment]
        self.Error(str(e))

    if interrupted:
        self.kernel_resp = {
            "status": "abort",
            "execution_count": self.execution_count,
        }

    exitcode, trace = self.check_exitcode()

    if exitcode:
        self.kernel_resp = {
            "status": "error",
            "execution_count": self.execution_count,
            "ename": "",
            "evalue": str(exitcode),
            "traceback": trace,
        }

    elif error:
        self.kernel_resp = {
            "status": "error",
            "execution_count": self.execution_count,
            "ename": "",
            "evalue": str(error),
            "traceback": str(tb),
        }

    else:
        self.kernel_resp = {
            "status": "ok",
            "execution_count": self.execution_count,
            "payload": [],
            "user_expressions": {},
        }

    if output:
        if stream_handler:
            stream_handler(output)
        else:
            return TextOutput(output)
    return None

check_exitcode() -> tuple[int, None]

Return (1, ["trace"]) if error.

Source code in metakernel/process_metakernel.py

def check_exitcode(self) -> tuple[int, None]:
    """
    Return (1, ["trace"]) if error.
    """
    return (0, None)

makeWrapper() -> REPLWrapper

In this method the REPLWrapper is created and returned. REPLWrapper takes the name of the executable, and arguments describing the executable prompt:

return REPLWrapper('bash', orig_prompt, prompt_change, prompt_cmd=prompt_cmd, extra_init_cmd=extra_init_cmd)

The parameters are:

:param orig_prompt: What the original prompt is (or is forced to be by prompt_cmd).

:param prompt_cmd: Used when the prompt is not printed by default (happens on Windows) to print something that we can search for.

:param prompt_change: Used to set the PS1/PS2 equivalents to something that is easier to tell apart from everything else, make sure it is a string with {0} and {1} in it for the PS1/PS2 fill-ins.

See metakernel.replwrap.REPLWrapper for more details.

Source code in metakernel/process_metakernel.py

def makeWrapper(self) -> REPLWrapper:
    """
    In this method the REPLWrapper is created and returned.
    REPLWrapper takes the name of the executable, and arguments
    describing the executable prompt:

    return REPLWrapper('bash', orig_prompt, prompt_change,
                       prompt_cmd=prompt_cmd,
                       extra_init_cmd=extra_init_cmd)

    The parameters are:

    :param orig_prompt: What the original prompt is (or is forced
    to be by prompt_cmd).

    :param prompt_cmd: Used when the prompt is not printed by default
    (happens on Windows) to print something that we can search for.

    :param prompt_change: Used to set the PS1/PS2 equivalents to
    something that is easier to tell apart from everything else,
    make sure it is a string with {0} and {1} in it for the
    PS1/PS2 fill-ins.

    See `metakernel.replwrap.REPLWrapper` for more details.

    """
    raise NotImplementedError

do_shutdown(restart: bool) -> dict[str, str] async

Shut down the app gracefully, saving history.

Source code in metakernel/process_metakernel.py

async def do_shutdown(self, restart: bool) -> dict[str, str]:
    """
    Shut down the app gracefully, saving history.
    """
    if self.wrapper is not None:
        try:
            self.wrapper.terminate()
        except Exception as e:
            self.Error(str(e))
    return await super().do_shutdown(restart)

restart_kernel() -> None

Restart the kernel

Source code in metakernel/process_metakernel.py

def restart_kernel(self) -> None:
    """Restart the kernel"""
    self.wrapper = self.makeWrapper()

REPLWrapper

metakernel.REPLWrapper

Wrapper for a REPL.

All prompts are interpreted as regexes. If you have special characters in the prompt, use re.escape to escape the characters.

:param cmd_or_spawn: This can either be an instance of :class:pexpect.spawn in which a REPL has already been started, or a str command to start a new REPL process. :param str prompt_regex: Regular expression representing process prompt, eg ">>>" in Python. :param str continuation_prompt_regex: Regular expression representing process continuation prompt, e.g. "..." in Python. :param str prompt_change_cmd: Optional kernel command that sets continuation-of-line-prompts, eg PS1 and PS2, such as "..." in Python. to something more unique. If this is None, the prompt will not be changed. This will be formatted with the new and continuation prompts as positional parameters, so you can use {} style formatting to insert them into the command. :param str new_prompt_regex: The more unique prompt to expect after the change. :param str stdin_prompt_regex: The regex for a stdin prompt from the child process. The prompt itself will be sent to the stdin_handler, so any sentinel value inserted will have to be removed by the caller. :param str extra_init_cmd: Commands to do extra initialisation, such as disabling pagers. :param str prompt_emit_cmd: Optional kernel command that emits the prompt when one is not emitted by default (typically happens on Windows only) :param bool force_prompt_on_continuation: Whether to force a prompt when we need to interrupt a continuation prompt. :param bool echo: Whether the child should echo, or in the case of Windows, whether the child does echo. :param dict extr_env: Extra env variables to send to the child. :param list args: Additional arguments to pass to the spawned command.

Source code in metakernel/replwrap.py

class REPLWrapper:
    """Wrapper for a REPL.

    All prompts are interpreted as regexes.  If you have special
    characters in the prompt, use `re.escape` to escape the characters.

    :param cmd_or_spawn: This can either be an instance of
    :class:`pexpect.spawn` in which a REPL has already been started,
    or a str command to start a new REPL process.
    :param str prompt_regex:  Regular expression representing process prompt, eg ">>>" in Python.
    :param str continuation_prompt_regex: Regular expression representing process continuation prompt, e.g. "..." in Python.
    :param str prompt_change_cmd: Optional kernel command that sets continuation-of-line-prompts, eg PS1 and PS2, such as "..." in Python.
        to something more unique. If this is ``None``, the prompt will not be
        changed. This will be formatted with the new and continuation prompts
        as positional parameters, so you can use ``{}`` style formatting to
        insert them into the command.
    :param str new_prompt_regex: The more unique prompt to expect after the change.
    :param str stdin_prompt_regex: The regex for a stdin prompt from the
        child process.  The prompt itself will be sent to the `stdin_handler`,
        so any sentinel value inserted will have to be removed by the caller.
    :param str extra_init_cmd: Commands to do extra initialisation, such as
      disabling pagers.
    :param str prompt_emit_cmd: Optional kernel command that emits the prompt
      when one is not emitted by default (typically happens on Windows only)
    :param bool force_prompt_on_continuation: Whether to force a prompt when
    we need to interrupt a continuation prompt.
    :param bool echo: Whether the child should echo, or in the case
    of Windows, whether the child does echo.
    :param dict extr_env: Extra env variables to send to the child.
    :param list args: Additional arguments to pass to the spawned command.
    """

    def __init__(
        self,
        cmd_or_spawn: Any,
        prompt_regex: Any,
        prompt_change_cmd: str | None,
        new_prompt_regex: Any = PEXPECT_PROMPT,
        continuation_prompt_regex: Any = PEXPECT_CONTINUATION_PROMPT,
        stdin_prompt_regex: Any = PEXPECT_STDIN_PROMPT,
        extra_init_cmd: str | None = None,
        prompt_emit_cmd: str | None = None,
        force_prompt_on_continuation: bool = False,
        echo: bool = False,
        extra_env: dict[str, Any] | None = None,
        encoding: str = "utf-8",
        args: list[str] | None = None,
    ) -> None:
        if isinstance(cmd_or_spawn, str):
            if extra_env:
                env = os.environ.copy()
                env.update(extra_env or {})
            else:
                env = None
            self.child = pexpect.spawnu(
                cmd_or_spawn,
                args=args,
                echo=echo,
                codec_errors="ignore",
                encoding=encoding,
                env=env,
            )
        else:
            self.child = cmd_or_spawn

        if self.child.echo and not echo:
            # Existing spawn instance has echo enabled, disable it
            # to prevent our input from being repeated to output.
            self.child.setecho(False)
            self.child.waitnoecho()

        self.echo = echo
        self.prompt_emit_cmd = prompt_emit_cmd
        self._force_prompt_on_continuation = force_prompt_on_continuation
        self.prompt_change_cmd: str | None = None
        self.prompt_regex: str | None = None

        if prompt_change_cmd is None:
            self.prompt_regex = prompt_regex
        else:
            formatted_prompt_change_cmd = prompt_change_cmd.format(
                new_prompt_regex, continuation_prompt_regex
            )
            self.set_prompt(prompt_regex, formatted_prompt_change_cmd)
            self.prompt_regex = new_prompt_regex
        self.continuation_prompt_regex = continuation_prompt_regex
        self.stdin_prompt_regex = stdin_prompt_regex

        self._stream_handler: Any = None
        self._stdin_handler: Any = None
        self._line_handler: Any = None

        self._expect_prompt()

        if extra_init_cmd is not None:
            self.run_command(extra_init_cmd)

        atexit.register(self.terminate)

    def sendline(self, line: str) -> None:
        self.child.sendline(line)
        if self.echo:
            self.child.readline()

    def set_prompt(self, prompt_regex: str, prompt_change_cmd: str) -> None:
        self.child.expect([r"[\s\S]+"])
        self.sendline(prompt_change_cmd)
        self.prompt_change_cmd = prompt_change_cmd

    def _expect_prompt(self, timeout: int | None = None) -> Any:
        """Expect a prompt from the child."""
        expects = [
            self.prompt_regex,
            self.continuation_prompt_regex,
            self.stdin_prompt_regex,
        ]
        if self.prompt_emit_cmd:
            self.sendline(self.prompt_emit_cmd)

        if self._stream_handler:
            return self._expect_prompt_stream(expects, timeout)

        if self._line_handler:
            expects += [self.child.crlf]

        while True:
            pos = self.child.expect(expects, timeout=timeout)
            # got a full prompt or continuation prompt.
            if pos in [0, 1]:
                return pos
            # got a stdin prompt
            if pos == 2:
                if not self._stdin_handler:
                    raise ValueError("Stdin Requested but not stdin handler available")

                resp = self._stdin_handler(self.child.after)
                self.sendline(resp)
            # got a newline
            else:
                self._line_handler(self.child.before.rstrip())

    def _expect_prompt_stream(
        self, expects: str | list[str | None], timeout: Any = None
    ) -> Any:
        """Expect a prompt with streaming output."""
        stream_handler = self._stream_handler
        stdin_handler = self._stdin_handler

        t0 = time.time()
        if timeout == -1:
            timeout = 30
        elif timeout is None:
            timeout = 1e6
        stream_timeout = 0.1

        unhandled_cr = False

        # Wait for a prompt, handling carriage returns.
        while True:
            if time.time() - t0 > timeout:
                raise pexpect.TIMEOUT("Timed out")

            try:
                # Wait for a prompt.
                pos = self.child.expect(expects, timeout=stream_timeout)
            except pexpect.TIMEOUT:
                # Process any carriage returns in the stream.
                while 1:
                    try:
                        self.child.expect(["\r"], timeout=0)
                        if unhandled_cr:
                            stream_handler("\r")
                        stream_handler(self.child.before)
                        unhandled_cr = True
                    except pexpect.TIMEOUT:
                        break
                continue

            # prompt or stdin request received, handle line.
            line = self.child.before

            # Handle cr state.
            if unhandled_cr:
                line = "\r" + line
            unhandled_cr = False

            # Handle stdin request.
            if pos == 2:
                if not stdin_handler:
                    raise ValueError("Stdin Requested but no stdin handler available")
                resp = stdin_handler(line + self.child.after)
                self.sendline(resp)
                continue

            # prompt received, but partial line precedes it
            if len(line) != 0:
                stream_handler(line)

            # exit on prompt received.
            break
        return pos

    def run_command(
        self,
        command: str,
        timeout: Any = None,
        stream_handler: Any = None,
        line_handler: Any = None,
        stdin_handler: Any = None,
    ) -> str:
        """Send a command to the REPL, wait for and return output.
        :param str command: The command to send. Trailing newlines are not needed.
          This should be a complete block of input that will trigger execution;
          if a continuation prompt is found after sending input, :exc:`ValueError`
          will be raised.
        :param int timeout: How long to wait for the next prompt. -1 means the
          default from the :class:`pexpect.spawn` object (default 30 seconds).
          None means to wait indefinitely.
        :param func stream_handler - A function that accepts a string to print as a streaming output
        :param func line_handler - A function that accepts a string to print as a line output
        :param stdin_handler - A function that prompts the user for input and
        returns a response.
        """
        # Split up multiline commands and feed them in bit-by-bit
        cmdlines = command.splitlines()
        # splitlines ignores trailing newlines - add it back in manually
        if command.endswith("\n"):
            cmdlines.append("")
        if not cmdlines:
            raise ValueError("No command was given")

        res = []
        self._line_handler = line_handler
        self._stream_handler = stream_handler
        self._stdin_handler = stdin_handler

        self.sendline(cmdlines[0])
        for line in cmdlines[1:]:
            if not self.prompt_emit_cmd:
                self._expect_prompt(timeout=timeout)
                res.append(strip_bracketing(self.child.before))
            self.sendline(line)

        # Command was fully submitted, now wait for the next prompt
        if self._expect_prompt(timeout=timeout) == 1:
            # Got a continuation prompt - could be a soft continuation (complete block
            # awaiting execution, e.g. `if True:\n    print('hi')`) or a hard
            # continuation (genuinely incomplete input, e.g. `if True:`).
            # Send an empty line to resolve soft continuations and check again.
            self.sendline("")
            if self._expect_prompt(timeout=timeout) == 1:
                # Still a continuation prompt - input was genuinely incomplete
                self.interrupt(continuation=True)
                raise ValueError(
                    "Continuation prompt found - input was incomplete:\n" + command
                )

        if self._stream_handler or self._line_handler:
            return ""
        return "".join(res + [strip_bracketing(self.child.before)])

    def interrupt(self, continuation: bool = False) -> Any:
        """Interrupt the process and wait for a prompt.

        Returns
        -------
        The value up to the prompt.
        """
        if pexpect.pty is not None:
            self.child.sendintr()
        else:
            self.child.kill(signal.SIGINT)  # type:ignore[unreachable]
        if continuation and self._force_prompt_on_continuation:
            self.sendline(self.prompt_change_cmd or "")
        while 1:
            try:
                self._expect_prompt(timeout=-1)
                break
            except KeyboardInterrupt:
                pass

        return self.child.before

    def terminate(self) -> Any:
        if pexpect.pty is not None:
            self.child.close()
            return self.child.terminate()
        try:  # type:ignore[unreachable]
            self.child.kill(signal.SIGTERM)
        except Exception as e:
            if e.errno != errno.EACCES:
                raise

Functions

run_command(command: str, timeout: Any = None, stream_handler: Any = None, line_handler: Any = None, stdin_handler: Any = None) -> str

Send a command to the REPL, wait for and return output. :param str command: The command to send. Trailing newlines are not needed. This should be a complete block of input that will trigger execution; if a continuation prompt is found after sending input, :exc:ValueError will be raised. :param int timeout: How long to wait for the next prompt. -1 means the default from the :class:pexpect.spawn object (default 30 seconds). None means to wait indefinitely. :param func stream_handler - A function that accepts a string to print as a streaming output :param func line_handler - A function that accepts a string to print as a line output :param stdin_handler - A function that prompts the user for input and returns a response.

Source code in metakernel/replwrap.py

def run_command(
    self,
    command: str,
    timeout: Any = None,
    stream_handler: Any = None,
    line_handler: Any = None,
    stdin_handler: Any = None,
) -> str:
    """Send a command to the REPL, wait for and return output.
    :param str command: The command to send. Trailing newlines are not needed.
      This should be a complete block of input that will trigger execution;
      if a continuation prompt is found after sending input, :exc:`ValueError`
      will be raised.
    :param int timeout: How long to wait for the next prompt. -1 means the
      default from the :class:`pexpect.spawn` object (default 30 seconds).
      None means to wait indefinitely.
    :param func stream_handler - A function that accepts a string to print as a streaming output
    :param func line_handler - A function that accepts a string to print as a line output
    :param stdin_handler - A function that prompts the user for input and
    returns a response.
    """
    # Split up multiline commands and feed them in bit-by-bit
    cmdlines = command.splitlines()
    # splitlines ignores trailing newlines - add it back in manually
    if command.endswith("\n"):
        cmdlines.append("")
    if not cmdlines:
        raise ValueError("No command was given")

    res = []
    self._line_handler = line_handler
    self._stream_handler = stream_handler
    self._stdin_handler = stdin_handler

    self.sendline(cmdlines[0])
    for line in cmdlines[1:]:
        if not self.prompt_emit_cmd:
            self._expect_prompt(timeout=timeout)
            res.append(strip_bracketing(self.child.before))
        self.sendline(line)

    # Command was fully submitted, now wait for the next prompt
    if self._expect_prompt(timeout=timeout) == 1:
        # Got a continuation prompt - could be a soft continuation (complete block
        # awaiting execution, e.g. `if True:\n    print('hi')`) or a hard
        # continuation (genuinely incomplete input, e.g. `if True:`).
        # Send an empty line to resolve soft continuations and check again.
        self.sendline("")
        if self._expect_prompt(timeout=timeout) == 1:
            # Still a continuation prompt - input was genuinely incomplete
            self.interrupt(continuation=True)
            raise ValueError(
                "Continuation prompt found - input was incomplete:\n" + command
            )

    if self._stream_handler or self._line_handler:
        return ""
    return "".join(res + [strip_bracketing(self.child.before)])

interrupt(continuation: bool = False) -> Any

Interrupt the process and wait for a prompt.

Returns:

Type	Description
The value up to the prompt.

Source code in metakernel/replwrap.py

def interrupt(self, continuation: bool = False) -> Any:
    """Interrupt the process and wait for a prompt.

    Returns
    -------
    The value up to the prompt.
    """
    if pexpect.pty is not None:
        self.child.sendintr()
    else:
        self.child.kill(signal.SIGINT)  # type:ignore[unreachable]
    if continuation and self._force_prompt_on_continuation:
        self.sendline(self.prompt_change_cmd or "")
    while 1:
        try:
            self._expect_prompt(timeout=-1)
            break
        except KeyboardInterrupt:
            pass

    return self.child.before