kernelspec

Add and remove kernel specifications for Jupyter.

Functions:

• make_argv –

  Returns an argument vector (argv) that can be used to start a Kernel.

• get_kernel_dir –

  The path to where kernel specs are stored for Jupyter.

• write_kernel_spec –

  Write a kernel spec for launching a kernel ref.

• validate_name –

  Check the name is a valid kernel name.

• get_kernel_info –

  Get a dic of kernels installed in kernel_dir.

• expand_path –

  Make the path absolute returning a new path object.

• import_launcher –

  Import a custom launcher or the default launcher.

Attributes:

• PROTOCOL_VERSION (str) –

  The protocol that is supported by the kernel.

• DEFAULT_LAUNCHER (str) –

  An importable path to the default interface to start the kernel.

• DEFAULT_COMMAND (tuple[str, ...]) –

PROTOCOL_VERSION module-attribute

PROTOCOL_VERSION: str = '5.5'

The protocol that is supported by the kernel.

DEFAULT_LAUNCHER module-attribute

DEFAULT_LAUNCHER: str = 'launch_interface'

An importable path to the default interface to start the kernel.

DEFAULT_COMMAND module-attribute

DEFAULT_COMMAND: tuple[str, ...] = (sys.executable, '-m', 'async_kernel', 'start')

Used by:

• Reference kernelspec
  •  make_argv
  •  write_kernel_spec

make_argv

make_argv(
    *,
    connection_file: str = "{connection_file}",
    name: str = "async",
    launcher: str | InterfaceStartType = "",
    command: tuple[str, ...] = DEFAULT_COMMAND,
    flags: Iterable[str] = (),
    **kwargs: Any,
) -> list[str]

Returns an argument vector (argv) that can be used to start a Kernel.

This function returns a list of arguments can be used directly start a kernel with subprocess.Popen. It will always call command.command_line as a python module.

Parameters:

• connection_file

  (str, default: '{connection_file}' ) –

  The path to the connection file.

• launcher

  (str | InterfaceStartType, default: '' ) –

  A self-contained function that accepts a dict of settings. Or as string import path to a callable responsible for launching the interface.

• name

  (str, default: 'async' ) –

  The name to use for the kernel.

• command

  (tuple[str, ...], default: DEFAULT_COMMAND ) –

  The command line command to call.

• flags

  (Iterable[str], default: () ) –

  Any number of flags to insert in argv. Flags will be prefixed with '--'.

• **kwargs

  (Any, default: {} ) –

  Additional settings to pass when creating the kernel passed to launcher.

Returns:

• list ( list[str] ) –

  A list of command-line arguments to launch the kernel module.

Referenced by:

• Reference interface  start_kernel_callable_interface

Source code in src/async_kernel/kernelspec.py

def make_argv(
    *,
    connection_file: str = "{connection_file}",
    name: str = "async",
    launcher: str | InterfaceStartType = "",
    command: tuple[str, ...] = DEFAULT_COMMAND,
    flags: Iterable[str] = (),
    **kwargs: Any,
) -> list[str]:
    """Returns an argument vector (argv) that can be used to start a `Kernel`.

    This function returns a list of arguments can be used directly start a kernel with [subprocess.Popen][].
    It will always call [command.command_line][] as a python module.

    Args:
        connection_file: The path to the connection file.
        launcher:
            A self-contained function that accepts a dict of settings.
            Or as string import path to a callable responsible for launching the interface.
        name: The name to use for the kernel.
        command: The command line command to call.
        flags: Any number of flags to insert in argv. Flags will be prefixed with '--'.
        **kwargs: Additional settings to pass when creating the kernel passed to `launcher`.

    Returns:
        list: A list of command-line arguments to launch the kernel module.
    """
    validate_name(name)
    argv = [*command, f"--connection_file={connection_file}", f"--name={name}"]
    if launcher:
        argv.append(f"--launcher={launcher}")
    for k, v in kwargs.items():
        argv.append(f"--{k}={v}")
    argv.extend(f"--{f.strip('-')}" for f in flags)
    return list(map(str, argv))

get_kernel_dir

get_kernel_dir(*, folder: str = '', prefix: str = '', user: bool = False) -> Path

The path to where kernel specs are stored for Jupyter.

If folder is passed, it is assumed to be the full path ending in 'kernels', prefix is ignored.

Parameters:

• folder

  (str, default: '' ) –

  The path to 'kernels' (must end with 'kernels').

• prefix

  (str, default: '' ) –

  Defaults to sys.prefix (installable for a particular environment).

• user

  (bool, default: False ) –

  Install for the user.

Search locations: https://jupyter-client.readthedocs.io/en/latest/kernels.html#kernel-specs

Source code in src/async_kernel/kernelspec.py

def get_kernel_dir(*, folder: str = "", prefix: str = "", user: bool = False) -> Path:
    """
    The path to where kernel specs are stored for Jupyter.

    If folder is passed, it is assumed to be the full path ending in 'kernels', prefix is ignored.

    Args:
        folder: The path to 'kernels' (must end with 'kernels').
        prefix: Defaults to sys.prefix (installable for a particular environment).
        user: Install for the user.

    Search locations: https://jupyter-client.readthedocs.io/en/latest/kernels.html#kernel-specs
    """

    if len([p for p in [folder, prefix, user] if p]) > 1:
        msg = "Providing more than one of [folder, prefix, user] is ambiguous"
        raise ValueError(msg)
    if user:
        from jupyter_core.paths import jupyter_data_dir  # noqa: PLC0415

        return Path(jupyter_data_dir()).joinpath("kernels")
    if folder:
        path = expand_path(folder)
        assert path.name.lower() == "kernels"
        return path
    return expand_path(prefix or sys.prefix).joinpath("share", "jupyter", "kernels")

write_kernel_spec

write_kernel_spec(
    *,
    path: Path | str | None = None,
    name: str = "async",
    display_name: str = "",
    user: bool = False,
    prefix: str = "",
    folder: str = "",
    launcher: str | InterfaceStartType = "",
    command: tuple[str, ...] = DEFAULT_COMMAND,
    connection_file: str = "{connection_file}",
    env: dict | None = None,
    metadata: dict | None = None,
    language="python",
    resources: Path | None = RESOURCES,
    flags: Iterable[str] = (),
    **kwargs: Any,
) -> Path

Write a kernel spec for launching a kernel ref.

Parameters:

• path

  (Path | str | None, default: None ) –

  The path where to write the spec.

• name

  (str, default: 'async' ) –

  The name of the kernel to use.

• display_name

  (str, default: '' ) –

  The display name for Jupyter to use for the kernel. The default is "Python ({name})".

• user

  (bool, default: False ) –

  To work with the user profile directory.

• prefix

  (str, default: '' ) –

  When provided the kernelspec will be installed to PREFIX/share/jupyter/kernels/KERNEL_NAME. This can be sys.prefix for installation inside virtual or conda envs.

• folder

  (str, default: '' ) –

  A direct path the the kernel spec folder (must end with a folder named 'kernels').

• launcher

  (str | InterfaceStartType, default: '' ) –

  The string import path to a callable that creates the Kernel or, a self-contained function that returns an instance of a Kernel.

• command

  (tuple[str, ...], default: DEFAULT_COMMAND ) –

  The command to execute to invoke the launcher.

• connection_file

  (str, default: '{connection_file}' ) –

  The path to the connection file.

• env

  (dict | None, default: None ) –

  A mapping environment variables for the kernel to set prior to starting.

• metadata

  (dict | None, default: None ) –

  A mapping of additional attributes to aid the client in kernel selection.

• resources

  (Path | None, default: RESOURCES ) –

  The path to the resources folder to include with the kernel spec.

• flags

  (Iterable[str], default: () ) –

  Flags to insert directly into the argv string.

• **kwargs

  (Any, default: {} ) –

  Pass additional settings to set on the instance of the Kernel when it is instantiated. Each setting should correspond to the dotted path to the attribute relative to the kernel. For example ..., **{'timeout'=0.1}).

Source code in src/async_kernel/kernelspec.py

def write_kernel_spec(
    *,
    path: Path | str | None = None,
    name: str = "async",
    display_name: str = "",
    user: bool = False,
    prefix: str = "",
    folder: str = "",
    launcher: str | InterfaceStartType = "",
    command: tuple[str, ...] = DEFAULT_COMMAND,
    connection_file: str = "{connection_file}",
    env: dict | None = None,
    metadata: dict | None = None,
    language="python",
    resources: Path | None = RESOURCES,
    flags: Iterable[str] = (),
    **kwargs: Any,
) -> Path:
    """
    Write a kernel spec for launching a kernel [ref](https://jupyter-client.readthedocs.io/en/stable/kernels.html#kernel-specs).

    Args:
        path:
            The path where to write the spec.
        name:
            The name of the kernel to use.
        display_name:
            The display name for Jupyter to use for the kernel. The default is `"Python ({name})"`.
        user:
            To work with the user profile directory.
        prefix:
            When provided the kernelspec will be installed to PREFIX/share/jupyter/kernels/KERNEL_NAME.
            This can be sys.prefix for installation inside virtual or conda envs.
        folder:
            A direct path the the kernel spec folder (must end with a folder named 'kernels').
        launcher:
            The string import path to a callable that creates the Kernel or, a *self-contained*
            function that returns an instance of a `Kernel`.
        command:
            The command to execute to invoke the launcher.
        connection_file:
            The path to the connection file.
        env:
            A mapping environment variables for the kernel to set prior to starting.
        metadata:
            A mapping of additional attributes to aid the client in kernel selection.
        resources:
            The path to the resources folder to include with the kernel spec.
        flags:
            Flags to insert directly into the argv string.
        **kwargs:
            Pass additional settings to set on the instance of the `Kernel` when it is instantiated.
            Each setting should correspond to the dotted path to the attribute relative to the kernel.
            For example `..., **{'timeout'=0.1})`.
    """
    import shutil  # noqa: PLC0415

    if path:
        path = expand_path(path)
    else:
        validate_name(name)
        path = get_kernel_dir(folder=folder, prefix=prefix, user=user).joinpath(name)

    if callable(launcher) and len(inspect.signature(launcher).parameters) != 1:
        msg = "Invalid signature! `launcher` must accept exactly one argument (a dict of settings)."
        raise ValueError(msg)
    # stage resources
    try:
        path.mkdir(parents=True, exist_ok=True)

        # launcher
        if callable(launcher):
            f = path.joinpath("launcher.py")
            f.write_text(textwrap.dedent(inspect.getsource(launcher)))
            launcher = f"{f}{CUSTOM_LAUNCHER_SEPARATOR}{launcher.__name__}"
        # validate
        if launcher and launcher != DEFAULT_LAUNCHER:
            assert len(inspect.signature(import_launcher(launcher)).parameters) == 1
        if resources:
            shutil.copytree(src=resources, dst=path, dirs_exist_ok=True)
        argv = make_argv(
            launcher=launcher,
            connection_file=connection_file,
            name=name,
            command=command,
            flags=flags,
            **kwargs,
        )
        spec: dict[str, list[Any] | Any | dict[Any, Any] | str | dict[str, bool]] = {
            "argv": argv,
            "env": env or {},
            "display_name": display_name or f"Python {sys.version.split()[0]} ({name})",
            "language": language,
            "interrupt_mode": "message",
            "metadata": metadata
            if metadata is not None
            else {"debugger": True, "concurrent": True, "supported_encryption": ["curve"]},
            "kernel_protocol_version": PROTOCOL_VERSION,
        }
        # write kernel.json
        path.joinpath("kernel.json").write_text(json.dumps(spec, indent=2))
    except Exception:
        shutil.rmtree(path, ignore_errors=True)
        raise
    else:
        return path

validate_name

validate_name(name: str) -> None

Check the name is a valid kernel name.

Raises:

• ValueError –

  If the name is not valid.

Source code in src/async_kernel/kernelspec.py

def validate_name(name: str, /) -> None:
    """
    Check the name is a valid kernel name.

    Raises:
        ValueError: If the name is not valid.
    """
    if not name or not re.match(re.compile(r"^[a-z0-9._\-]+$", re.IGNORECASE), name):
        msg = f"Invalid {name=}!"
        raise ValueError(msg)

get_kernel_info

get_kernel_info(kernel_dir: Path) -> dict[str, dict[str, Any]]

Get a dic of kernels installed in kernel_dir.

Source code in src/async_kernel/kernelspec.py

def get_kernel_info(kernel_dir: Path) -> dict[str, dict[str, Any]]:
    """Get a dic of kernels installed in kernel_dir."""
    kernels = {}
    if kernel_dir.is_dir():
        for path in kernel_dir.iterdir():
            if path.is_dir() and (info_file := path.joinpath("kernel.json")).exists():
                kernels[path.name] = json.loads(info_file.read_bytes())
    return kernels

expand_path

expand_path(path: str | Path) -> Path

Make the path absolute returning a new path object.

Parameters:

• path

  (str | Path) –

  The path to process.

Substitutions

• Windows environment variables are accepted such as %APPDATA% and %PROGRAMDATA%. -~ is also substituted with pathlib.Path.expanduser.

Source code in src/async_kernel/kernelspec.py

def expand_path(path: str | Path) -> Path:
    """
    Make the path absolute returning a new path object.

    Args:
        path: The path to process.

    Substitutions:
        - Windows environment variables are accepted such as `%APPDATA%` and `%PROGRAMDATA%`.
        -`~` is also substituted with [pathlib.Path.expanduser][].
    """
    if sys.platform == "win32" and str(path).startswith("%"):
        key, base = str(path).split("%")[1:3]
        key = key.strip("%").upper()
        pth_ = Path(os.environ[key])
        assert pth_.exists()
        pth = pth_.joinpath(base.strip("\\/"))
        assert pth.parts[0] == pth_.parts[0]
        return pth
    if isinstance(path, str):
        path = Path(path)
    return path.expanduser().absolute()

import_launcher

import_launcher(launcher: str = '') -> InterfaceStartType

Import a custom launcher or the default launcher.

Parameters:

• launcher

  (str, default: '' ) –

  The name of the interface factory.

Returns:

• callable ( InterfaceStartType ) –

  The imported function responsible for launching the interface.

Referenced by:

• Reference interface
  •  BaseInterface
  •  BaseInterface  launcher

Source code in src/async_kernel/kernelspec.py

def import_launcher(launcher: str = "", /) -> InterfaceStartType:
    """
    Import a custom launcher or the default launcher.

    Args:
        launcher: The name of the interface factory.

    Returns:
        callable: The imported function responsible for launching the interface.
    """
    if CUSTOM_LAUNCHER_SEPARATOR in launcher:
        name, factory_name = launcher.split(CUSTOM_LAUNCHER_SEPARATOR)
        glbls = {}
        exec(Path(name).read_bytes(), glbls)
        return glbls[factory_name]
    from async_kernel.common import import_item  # noqa: PLC0415

    if not launcher or launcher == DEFAULT_LAUNCHER:
        launcher = "async_kernel.interface.launch_interface"

    return import_item(launcher)