Skip to content

kernelspec

Add and remove kernel specifications for Jupyter.

Functions:

make_argv 

make_argv(
    *,
    connection_file: str = "{connection_file}",
    kernel_name: str = "async",
    start_interface: str | InterfaceStartType = DEFAULT_START_INTERFACE,
    fullpath: bool = True,
    **kwargs: dict[str, 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.

  • start_interface

    (str | InterfaceStartType, default: DEFAULT_START_INTERFACE ) –

    Either the kernel factory object itself, or the string import path to a callable that returns a non-started kernel.

  • kernel_name

    (str, default: 'async' ) –

    The name of the kernel to use.

  • fullpath

    (bool, default: True ) –

    If True the full path to the executable is used, otherwise 'python' is used.

  • **kwargs

    (dict[str, Any], default: {} ) –

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

Returns:

  • list ( list[str] ) –

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

Source code in src/async_kernel/kernelspec.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def make_argv(
    *,
    connection_file: str = "{connection_file}",
    kernel_name: str = "async",
    start_interface: str | InterfaceStartType = DEFAULT_START_INTERFACE,
    fullpath: bool = True,
    **kwargs: dict[str, 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.
        start_interface: Either the kernel factory object itself, or the string import path to a
            callable that returns a non-started kernel.
        kernel_name: The name of the kernel to use.
        fullpath: If True the full path to the executable is used, otherwise 'python' is used.
        **kwargs: Additional settings to pass when creating the kernel passed to `start_interface`.

    Returns:
        list: A list of command-line arguments to launch the kernel module.
    """
    argv = [(sys.executable if fullpath else "python"), "-m", "async_kernel", "-f", connection_file]
    for k, v in ({"start_interface": start_interface, "kernel_name": kernel_name} | kwargs).items():
        argv.append(f"--{k}={v}")
    return argv

write_kernel_spec 

write_kernel_spec(
    path: Path | str | None = None,
    *,
    kernel_name: str = "async",
    display_name: str = "",
    fullpath: bool = False,
    prefix: str = "",
    start_interface: str | InterfaceStartType = DEFAULT_START_INTERFACE,
    connection_file: str = "{connection_file}",
    env: dict | None = None,
    metadata: dict | None = None,
    **kwargs: dict[str, 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.

  • kernel_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 ({kernel_name})".

  • fullpath

    (bool, default: False ) –

    If True the full path to the executable is used, otherwise 'python' is used.

  • prefix

    (str, default: '' ) –

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

  • start_interface

    (str | InterfaceStartType, default: DEFAULT_START_INTERFACE ) –

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

  • 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.

  • **kwargs

    (dict[str, 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 ..., **{'shell.timeout'=0.1}).

Example passing a callable start_interface:

When `start_interface` is passed as a callable, the callable is stored in the file
'kernel_spec.py' inside the kernelspec folder.

```python
import async_kernel.kernelspec


def start_interface(settings):
    from async_kernel import Kernel

    class MyKernel(Kernel):
        async def execute_request(self, job):
            print(job)
            return await super().execute_request(job)

    return MyKernel(settings)


async_kernel.kernelspec.write_kernel_spec(
    kernel_name="async-print-job", start_interface=start_interface
)
```

Warning:
    Moving the spec folder will break the import which is stored as an absolute path.
Referenced by:
Source code in src/async_kernel/kernelspec.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def write_kernel_spec(
    path: Path | str | None = None,
    *,
    kernel_name: str = "async",
    display_name: str = "",
    fullpath: bool = False,
    prefix: str = "",
    start_interface: str | InterfaceStartType = DEFAULT_START_INTERFACE,
    connection_file: str = "{connection_file}",
    env: dict | None = None,
    metadata: dict | None = None,
    **kwargs: dict[str, 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.
        kernel_name: The name of the kernel to use.
        display_name: The display name for Jupyter to use for the kernel. The default is `"Python ({kernel_name})"`.
        fullpath: If True the full path to the executable is used, otherwise 'python' is used.
        prefix: given, the kernelspec will be installed to PREFIX/share/jupyter/kernels/KERNEL_NAME.
            This can be sys.prefix for installation inside virtual or conda envs.
        start_interface: The string import path to a callable that creates the Kernel or,
            a *self-contained* function that returns an instance of a `Kernel`.
        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.
        **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 `..., **{'shell.timeout'=0.1})`.

    Example passing a callable start_interface:

        When `start_interface` is passed as a callable, the callable is stored in the file
        'kernel_spec.py' inside the kernelspec folder.

        ```python
        import async_kernel.kernelspec


        def start_interface(settings):
            from async_kernel import Kernel

            class MyKernel(Kernel):
                async def execute_request(self, job):
                    print(job)
                    return await super().execute_request(job)

            return MyKernel(settings)


        async_kernel.kernelspec.write_kernel_spec(
            kernel_name="async-print-job", start_interface=start_interface
        )
        ```

        Warning:
            Moving the spec folder will break the import which is stored as an absolute path.
    """

    assert re.match(re.compile(r"^[a-z0-9._\-]+$", re.IGNORECASE), kernel_name)
    path = Path(path) if path else (get_kernel_dir(prefix) / kernel_name)
    # stage resources
    try:
        path.mkdir(parents=True, exist_ok=True)
        if callable(start_interface):
            path.joinpath("start_interface.py").write_text(textwrap.dedent(inspect.getsource(start_interface)))
            start_interface = f"{path}{CUSTOM_KERNEL_MARKER}{start_interface.__name__}"
        # validate
        if start_interface != DEFAULT_START_INTERFACE:
            import_start_interface(start_interface)
        shutil.copytree(src=RESOURCES, dst=path, dirs_exist_ok=True)

        argv = make_argv(
            start_interface=start_interface,
            connection_file=connection_file,
            kernel_name=kernel_name,
            fullpath=fullpath,
            **kwargs,
        )
        spec = {
            "argv": argv,
            "env": env or {},
            "display_name": display_name or f"Python ({kernel_name})",
            "language": "python",
            "interrupt_mode": "message",
            "metadata": metadata if metadata is not None else {"debugger": True, "concurrent": True},
            "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

get_kernel_dir 

get_kernel_dir(prefix: str = '') -> Path

The path to where kernel specs are stored for Jupyter.

Parameters:

  • prefix

    (str, default: '' ) –

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

Source code in src/async_kernel/kernelspec.py 
166
167
168
169
170
171
172
173
def get_kernel_dir(prefix: str = "") -> Path:
    """
    The path to where kernel specs are stored for Jupyter.

    Args:
        prefix: Defaults to sys.prefix (installable for a particular environment).
    """
    return Path(prefix or sys.prefix) / "share/jupyter/kernels"

import_start_interface 

import_start_interface(start_interface: str = '') -> InterfaceStartType

Import the kernel interface starter as defined in a kernel spec.

Parameters:

  • start_interface

    (str, default: '' ) –

    The name of the interface factory.

Returns:

  • InterfaceStartType –

    The kernel factory.

Source code in src/async_kernel/kernelspec.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def import_start_interface(start_interface: str = "", /) -> InterfaceStartType:
    """
    Import the kernel interface starter as defined in a kernel spec.

    Args:
        start_interface: The name of the interface factory.

    Returns:
        The kernel factory.
    """

    if CUSTOM_KERNEL_MARKER in start_interface:
        path, factory_name = start_interface.split(CUSTOM_KERNEL_MARKER)
        try:
            sys.path.insert(0, path)
            import start_interface as kf  # noqa: PLC0415

            factory = getattr(kf, factory_name)
            assert len(inspect.signature(factory).parameters) == 1
            return factory
        finally:
            sys.path.remove(path)
    from async_kernel.common import import_item  # noqa: PLC0415

    return import_item(start_interface or DEFAULT_START_INTERFACE)