Skip to content

typing

Provides publicly available typing definitions.

Classes:

Attributes:

  • NoValue (Final) –

    A sentinel to indicate a value has not been provided.

  • DebugMessage –

    A TypeAlias for a debug message.

  • Content –

    A TypeAlias for the content in Message.

  • HandlerType –

    A TypeAlias for the handler of message requests.

NoValue module-attribute 

NoValue: Final = Sentinel('NoValue')

A sentinel to indicate a value has not been provided.

Used by:

DebugMessage module-attribute 

DebugMessage = dict[str, Any]

A TypeAlias for a debug message.

Used by:

Content module-attribute 

Content = dict[str, Any]

A TypeAlias for the content in Message.

Notes
  • The content of a message handler can provide 'buffers'. When present, the buffers are extracted from dict and handled separately by the interface.
Returned by: Used by:

HandlerType module-attribute 

HandlerType = Callable[[Job], Awaitable[Content | None]]

A TypeAlias for the handler of message requests.

Backend

Bases: StrEnum

An enum of library names corresponding to anyio.

Referenced by: Returned by: Used by:

Attributes:

  • asyncio –

    An asyncio style event loop.

  • trio –

    A trio style event loop.

Source code in src/async_kernel/typing.py 
58
59
60
61
62
63
64
65
class Backend(enum.StrEnum):
    """An enum of library names corresponding to anyio."""

    asyncio = "asyncio"
    """An asyncio style event loop."""

    trio = "trio"
    """A trio style event loop."""

asyncio class-attribute instance-attribute 

asyncio = 'asyncio'

An asyncio style event loop.

Referenced by:

trio class-attribute instance-attribute 

trio = 'trio'

A trio style event loop.

Hosts

Bases: StrEnum

An enum of host names that are available in async_kernel.event_loop.run.run.

Referenced by: Returned by: Used by:

Methods:

  • from_gui –

    Transform a matplotlib gui type to a host name if possible.

Attributes:

Source code in src/async_kernel/typing.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@final
class Hosts(enum.StrEnum):
    """An enum of host names that are available in [async_kernel.event_loop.run.run][]."""

    tk = "tk"
    """An eventloop for [tkinter][]."""

    qt = "qt"
    """An event loop for [Qt](https://wiki.qt.io/Qt_for_Python)."""

    custom = "custom"
    """A custom host."""

    @classmethod
    def from_gui(cls, gui: str | None, /) -> Hosts | None:
        """Transform a matplotlib gui type to a host name if possible."""
        if gui:
            if gui == "tk":
                return Hosts.tk
            if gui == "qt":
                return Hosts.qt
        return None

tk class-attribute instance-attribute 

tk = 'tk'

An eventloop for tkinter.

qt class-attribute instance-attribute 

qt = 'qt'

An event loop for Qt.

custom class-attribute instance-attribute 

custom = 'custom'

A custom host.

from_gui classmethod 

from_gui(gui: str | None) -> Hosts | None

Transform a matplotlib gui type to a host name if possible.

Source code in src/async_kernel/typing.py 
81
82
83
84
85
86
87
88
89
@classmethod
def from_gui(cls, gui: str | None, /) -> Hosts | None:
    """Transform a matplotlib gui type to a host name if possible."""
    if gui:
        if gui == "tk":
            return Hosts.tk
        if gui == "qt":
            return Hosts.qt
    return None

Channel

Bases: StrEnum

An enum of channel names Ref.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class Channel(enum.StrEnum):
    """An enum of channel names [Ref](https://jupyter-client.readthedocs.io/en/stable/messaging.html#introduction)."""

    heartbeat = "hb"
    ""
    shell = "shell"
    ""
    stdin = "stdin"
    ""
    control = "control"
    ""
    iopub = "iopub"
    ""

heartbeat class-attribute instance-attribute 

heartbeat = 'hb'

shell class-attribute instance-attribute 

shell = 'shell'
Returned by: Used by:

stdin class-attribute instance-attribute 

stdin = 'stdin'

control class-attribute instance-attribute 

control = 'control'
Returned by: Used by:

iopub class-attribute instance-attribute 

iopub = 'iopub'

RunMode

Bases: StrEnum

An Enum of the run modes available for handling execute_request messages.

Cell overrides

The user can also specify an execution mode in execute requests.

Top line comment: 

# task
or

```python
# thread
```

Methods:

  • to_runmode –

    Converts value to Runmode, CallerCreateOptions or default where it is not possible.

  • line_to_options –

    Convert the line string to CallerCreateOptions.

Attributes:

Source code in src/async_kernel/typing.py 
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class RunMode(enum.StrEnum):
    """An Enum of the run modes available for handling `execute_request` messages.

    Cell overrides:
        The user can also specify an execution mode in execute requests.

        Top line comment:
            ```python
            # task
            ```
            or

            ```python
            # thread
            ```
    """

    @override
    def __str__(self):
        return f"# {self.name}"

    @override
    def __eq__(self, value: object, /) -> bool:
        return str(value) in {self.name, str(self), repr(self)}

    @override
    def __hash__(self) -> int:
        return hash(self.name)

    @classmethod
    def to_runmode(cls, value: Any, default: T = None, /) -> Self | T | CallerCreateOptions:
        """Converts value to `Runmode`, `CallerCreateOptions` or  default where it is not possible.

        `CallerCreateOptions` will only return when a string is passed specifying a thread and options.

        Example:
            `value = "# thread name='My exeecutor' backend=trio`
        """
        try:
            return cls(value)
        except ValueError:
            if isinstance(value, str):
                if value.startswith(("# ", "##")):
                    return cls.to_runmode(value[2:].strip(), default)
                if value.startswith("thread"):
                    return cls.line_to_options(value.removeprefix("thread"))
            return default

    @classmethod
    def line_to_options(cls, line: str, /) -> CallerCreateOptions:
        """Convert the line string to CallerCreateOptions."""
        import shlex  # noqa: PLC0415

        items = CallerCreateOptions()
        for v in shlex.split(line):
            k, val = v.split("=", maxsplit=1)
            try:
                items[k] = eval(val)
            except Exception:
                items[k] = val
        if items and not items.get("name"):
            msg = "'name' must be specified when providing settings!"
            raise ValueError(msg)
        if invalid := set(items).difference(get_annotations(CallerCreateOptions)):
            msg = f"One or more invalid options found! valid={list(get_annotations(CallerCreateOptions))} invalid={list(invalid)}"
            raise ValueError(msg)
        return items

    queue = "queue"
    """Run the message handler using [async_kernel.caller.Caller.queue_call][]."""

    task = "task"
    """Run the message handler using [async_kernel.caller.Caller.call_soon][]."""

    thread = "thread"
    """Run the message handler using [async_kernel.caller.Caller.to_thread][]."""

queue class-attribute instance-attribute 

queue = 'queue'

Run the message handler using async_kernel.caller.Caller.queue_call.

task class-attribute instance-attribute 

task = 'task'

Run the message handler using async_kernel.caller.Caller.call_soon.

thread class-attribute instance-attribute 

thread = 'thread'

Run the message handler using async_kernel.caller.Caller.to_thread.

to_runmode classmethod 

to_runmode(value: Any, default: T = None) -> Self | T | CallerCreateOptions

Converts value to Runmode, CallerCreateOptions or default where it is not possible.

CallerCreateOptions will only return when a string is passed specifying a thread and options.

Example

value = "# thread name='My exeecutor' backend=trio

Source code in src/async_kernel/typing.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@classmethod
def to_runmode(cls, value: Any, default: T = None, /) -> Self | T | CallerCreateOptions:
    """Converts value to `Runmode`, `CallerCreateOptions` or  default where it is not possible.

    `CallerCreateOptions` will only return when a string is passed specifying a thread and options.

    Example:
        `value = "# thread name='My exeecutor' backend=trio`
    """
    try:
        return cls(value)
    except ValueError:
        if isinstance(value, str):
            if value.startswith(("# ", "##")):
                return cls.to_runmode(value[2:].strip(), default)
            if value.startswith("thread"):
                return cls.line_to_options(value.removeprefix("thread"))
        return default

line_to_options classmethod 

line_to_options(line: str) -> CallerCreateOptions

Convert the line string to CallerCreateOptions.

Source code in src/async_kernel/typing.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@classmethod
def line_to_options(cls, line: str, /) -> CallerCreateOptions:
    """Convert the line string to CallerCreateOptions."""
    import shlex  # noqa: PLC0415

    items = CallerCreateOptions()
    for v in shlex.split(line):
        k, val = v.split("=", maxsplit=1)
        try:
            items[k] = eval(val)
        except Exception:
            items[k] = val
    if items and not items.get("name"):
        msg = "'name' must be specified when providing settings!"
        raise ValueError(msg)
    if invalid := set(items).difference(get_annotations(CallerCreateOptions)):
        msg = f"One or more invalid options found! valid={list(get_annotations(CallerCreateOptions))} invalid={list(invalid)}"
        raise ValueError(msg)
    return items

MsgType

Bases: StrEnum

An enumeration of Message msg_type for shell and control messages.

Some message types are on the control channel only.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
class MsgType(enum.StrEnum):
    """An enumeration of Message `msg_type` for [shell and control messages]( https://jupyter-client.readthedocs.io/en/stable/messaging.html#messages-on-the-shell-router-dealer-channel).

    Some message types are on the [control channel](https://jupyter-client.readthedocs.io/en/stable/messaging.html#messages-on-the-control-router-dealer-channel) only.
    """

    kernel_info_request = "kernel_info_request"
    """[async_kernel.kernel.Kernel.kernel_info_request][]"""

    comm_info_request = "comm_info_request"
    """[async_kernel.kernel.Kernel.comm_info_request][]"""

    execute_request = "execute_request"
    """[async_kernel.kernel.Kernel.execute_request][]"""

    complete_request = "complete_request"
    """[async_kernel.kernel.Kernel.complete_request][]"""

    is_complete_request = "is_complete_request"
    """[async_kernel.kernel.Kernel.is_complete_request][]"""

    inspect_request = "inspect_request"
    """[async_kernel.kernel.Kernel.inspect_request][]"""

    history_request = "history_request"
    """[async_kernel.kernel.Kernel.history_request][]"""

    comm_open = "comm_open"
    """[async_kernel.kernel.Kernel.comm_open][]"""

    comm_msg = "comm_msg"
    """[async_kernel.kernel.Kernel.comm_msg][]"""

    comm_close = "comm_close"
    """[async_kernel.kernel.Kernel.comm_close][]"""

    # Control
    interrupt_request = "interrupt_request"
    """[async_kernel.kernel.Kernel.interrupt_request][] (control channel only)"""

    shutdown_request = "shutdown_request"
    """[async_kernel.kernel.Kernel.shutdown_request][] (control channel only)"""

    debug_request = "debug_request"
    """[async_kernel.kernel.Kernel.debug_request][] (control channel only)"""

    create_subshell_request = "create_subshell_request"
    """[async_kernel.kernel.Kernel.create_subshell_request][] (control channel only)"""

    delete_subshell_request = "delete_subshell_request"
    """[async_kernel.kernel.Kernel.delete_subshell_request][] (control channel only)"""

    list_subshell_request = "list_subshell_request"
    """[async_kernel.kernel.Kernel.debug_request][] (control channel only)"""

    # Reverse messaging (kernel -> client)
    input_request = "input_request"
    """A message sent from the kernel interface to a client asking for raw input."""

    # iopub messages (kernel -> client

    iopub_welcome = "iopub_welcome"
    """A welcome message on the iopub channel for new iopub channel subscriptions."""

    iopub_status = "status"
    """An iopub message about a handlers status which can be 'busy' or 'idle'."""

    iopub_execute_input = "execute_input"
    """An iopub message with detail of an execute request."""

    iopub_execute_result = "execute_result"
    """An iopub message with for the global display hook. Generally the last executed line of an execute request."""

    iopub_error = "error"
    """An iopub message for an error."""

    iopub_stream = "stream"
    """Stream data such as stdout and stderr."""

    iopub_display_data = "display_data"
    """An iopub message with display output data."""

    iopub_update_display_data = "iopub_update_display_data"
    """An iopub message to update display data."""

    iopub_clear_output = "clear_output"
    """An iopub display message instructing the associated display to clear."""

    debug_event = "debug_event"
    """An event published by an attached debug adapter."""

    # Reply messages (kernel -> client)
    kernel_info_reply = "kernel_info_reply"

    comm_info_reply = "comm_info_reply"

    execute_reply = "execute_reply"

    complete_reply = "complete_reply"

    is_complete_reply = "is_complete_reply"

    inspect_reply = "inspect_reply"

    history_reply = "history_reply"

    interrupt_reply = "interrupt_reply"

    shutdown_reply = "shutdown_reply"

    debug_reply = "debug_reply"

    create_subshell_reply = "create_subshell_reply"

    delete_subshell_reply = "delete_subshell_reply"

    list_subshell_reply = "list_subshell_reply"

    # Reverse reply (client -> kernel)
    input_reply = "input_reply"
    """A reply sent from the client to a kernel interface corresponding to an input request."""

kernel_info_request class-attribute instance-attribute 

kernel_info_request = 'kernel_info_request'

async_kernel.kernel.Kernel.kernel_info_request

comm_info_request class-attribute instance-attribute 

comm_info_request = 'comm_info_request'

async_kernel.kernel.Kernel.comm_info_request

execute_request class-attribute instance-attribute 

execute_request = 'execute_request'

async_kernel.kernel.Kernel.execute_request

complete_request class-attribute instance-attribute 

complete_request = 'complete_request'

async_kernel.kernel.Kernel.complete_request

is_complete_request class-attribute instance-attribute 

is_complete_request = 'is_complete_request'

async_kernel.kernel.Kernel.is_complete_request

inspect_request class-attribute instance-attribute 

inspect_request = 'inspect_request'

async_kernel.kernel.Kernel.inspect_request

history_request class-attribute instance-attribute 

history_request = 'history_request'

async_kernel.kernel.Kernel.history_request

comm_open class-attribute instance-attribute 

comm_open = 'comm_open'

async_kernel.kernel.Kernel.comm_open

Used by:

comm_msg class-attribute instance-attribute 

comm_msg = 'comm_msg'

async_kernel.kernel.Kernel.comm_msg

Used by:

comm_close class-attribute instance-attribute 

comm_close = 'comm_close'

async_kernel.kernel.Kernel.comm_close

Used by:

interrupt_request class-attribute instance-attribute 

interrupt_request = 'interrupt_request'

async_kernel.kernel.Kernel.interrupt_request (control channel only)

shutdown_request class-attribute instance-attribute 

shutdown_request = 'shutdown_request'

async_kernel.kernel.Kernel.shutdown_request (control channel only)

debug_request class-attribute instance-attribute 

debug_request = 'debug_request'

async_kernel.kernel.Kernel.debug_request (control channel only)

create_subshell_request class-attribute instance-attribute 

create_subshell_request = 'create_subshell_request'

async_kernel.kernel.Kernel.create_subshell_request (control channel only)

delete_subshell_request class-attribute instance-attribute 

delete_subshell_request = 'delete_subshell_request'

async_kernel.kernel.Kernel.delete_subshell_request (control channel only)

list_subshell_request class-attribute instance-attribute 

list_subshell_request = 'list_subshell_request'

async_kernel.kernel.Kernel.debug_request (control channel only)

input_request class-attribute instance-attribute 

input_request = 'input_request'

A message sent from the kernel interface to a client asking for raw input.

iopub_welcome class-attribute instance-attribute 

iopub_welcome = 'iopub_welcome'

A welcome message on the iopub channel for new iopub channel subscriptions.

iopub_status class-attribute instance-attribute 

iopub_status = 'status'

An iopub message about a handlers status which can be 'busy' or 'idle'.

iopub_execute_input class-attribute instance-attribute 

iopub_execute_input = 'execute_input'

An iopub message with detail of an execute request.

iopub_execute_result class-attribute instance-attribute 

iopub_execute_result = 'execute_result'

An iopub message with for the global display hook. Generally the last executed line of an execute request.

iopub_error class-attribute instance-attribute 

iopub_error = 'error'

An iopub message for an error.

iopub_stream class-attribute instance-attribute 

iopub_stream = 'stream'

Stream data such as stdout and stderr.

iopub_display_data class-attribute instance-attribute 

iopub_display_data = 'display_data'

An iopub message with display output data.

iopub_update_display_data class-attribute instance-attribute 

iopub_update_display_data = 'iopub_update_display_data'

An iopub message to update display data.

iopub_clear_output class-attribute instance-attribute 

iopub_clear_output = 'clear_output'

An iopub display message instructing the associated display to clear.

debug_event class-attribute instance-attribute 

debug_event = 'debug_event'

An event published by an attached debug adapter.

input_reply class-attribute instance-attribute 

input_reply = 'input_reply'

A reply sent from the client to a kernel interface corresponding to an input request.

Tags

Bases: StrEnum

Cell tags used in async-kernel.

Info

Tags are can be added per cell.

Used by:

Methods:

  • get_value –

    Extract the value and convert it to match the type of default using tag=value.

Attributes:

  • raises_exception –

    Indicates the cell should expect an exception to be raised.

  • stop_on_error –

    Override stop_on_error.

  • timeout –

    Specify a timeout in seconds for code execution to complete.

Source code in src/async_kernel/typing.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
class Tags(enum.StrEnum):
    """Cell tags used in async-kernel.

    Info:
        Tags are can be added per cell.

        - Jupyter: via the [right side bar](https://jupyterlab.readthedocs.io/en/stable/user/interface.html#left-and-right-sidebar).
        - VS Code: via [Jupyter variables explorer](https://code.visualstudio.com/docs/python/jupyter-support-py#_variables-explorer-and-data-viewer)
    """

    @override
    def __eq__(self, value: object, /) -> bool:
        return str(value).replace("-", "_").split("=")[0] == self.name

    @override
    def __hash__(self) -> int:
        return hash(self.name)

    @classmethod
    def get_value(cls, value: str, default: T_fsb) -> T_fsb:
        """Extract the value and convert it to match the type of default using `tag=value`."""
        try:
            value = value.split("=")[1].strip()
        except Exception:
            return default
        try:
            if isinstance(default, bool):
                return value.lower() == "true"
            if isinstance(default, int):
                return int(value)
            if isinstance(default, float):
                return float(value)
            if isinstance(default, str):  # pyright: ignore[reportUnnecessaryIsInstance]
                return value
        except Exception:
            return default
        msg = "default must be a one of string, int, float, bool."  # pyright: ignore[reportUnreachable]
        raise ValueError(msg)

    raises_exception = "raises-exception"
    """Indicates the cell should expect an exception to be raised. 

    Notes:
        - When an exception is raised, stop_on_error is False/
        - When an exception is **not** raised an exception will be raise and stop_on_error is True.
    """

    stop_on_error = "stop-on-error"
    """
    Override `stop_on_error`.

    Examples:

        - True
            - stop_on_error=true
            - stop_on_error=True
        - False
            - stop_on_error=False
    """

    timeout = "timeout"
    """
    Specify a timeout in seconds for code execution to complete.

    Examples:

        - timeout=0.0 (no timeout)
        - timeout=0.1 (100 ms)
    """

raises_exception class-attribute instance-attribute 

raises_exception = 'raises-exception'

Indicates the cell should expect an exception to be raised.

Notes
  • When an exception is raised, stop_on_error is False/
  • When an exception is not raised an exception will be raise and stop_on_error is True.

stop_on_error class-attribute instance-attribute 

stop_on_error = 'stop-on-error'

Override stop_on_error.

Examples:

- True
    - stop_on_error=true
    - stop_on_error=True
- False
    - stop_on_error=False

timeout class-attribute instance-attribute 

timeout = 'timeout'

Specify a timeout in seconds for code execution to complete.

Examples:

- timeout=0.0 (no timeout)
- timeout=0.1 (100 ms)

get_value classmethod 

get_value(value: str, default: T_fsb) -> T_fsb

Extract the value and convert it to match the type of default using tag=value.

Source code in src/async_kernel/typing.py 
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
@classmethod
def get_value(cls, value: str, default: T_fsb) -> T_fsb:
    """Extract the value and convert it to match the type of default using `tag=value`."""
    try:
        value = value.split("=")[1].strip()
    except Exception:
        return default
    try:
        if isinstance(default, bool):
            return value.lower() == "true"
        if isinstance(default, int):
            return int(value)
        if isinstance(default, float):
            return float(value)
        if isinstance(default, str):  # pyright: ignore[reportUnnecessaryIsInstance]
            return value
    except Exception:
        return default
    msg = "default must be a one of string, int, float, bool."  # pyright: ignore[reportUnreachable]
    raise ValueError(msg)

CallerState

Bases: Enum

The State of a async_kernel.caller.Caller.

Source code in src/async_kernel/typing.py 
400
401
402
403
404
405
406
407
408
class CallerState(enum.Enum):
    """The State of a [async_kernel.caller.Caller][]."""

    initial = enum.auto()
    start_sync = enum.auto()
    starting = enum.auto()
    running = enum.auto()
    stopping = enum.auto()
    stopped = enum.auto()

MsgHeader

Bases: TypedDict

A message header.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
class MsgHeader(TypedDict):
    """A [message header](https://jupyter-client.readthedocs.io/en/stable/messaging.html#message-header)."""

    msg_id: str
    ""
    session: str
    ""
    username: str
    ""
    date: str | datetime.datetime
    ""
    msg_type: MsgType | str
    ""
    version: str
    ""
    subshell_id: NotRequired[str | None]
    ""

msg_id instance-attribute 

msg_id: str

session instance-attribute 

session: str

username instance-attribute 

username: str

date instance-attribute 

date: str | datetime

msg_type instance-attribute 

msg_type: MsgType | str

version instance-attribute 

version: str

subshell_id instance-attribute 

subshell_id: NotRequired[str | None]

Message

Bases: TypedDict, Generic[T]

A message.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
class Message(TypedDict, Generic[T]):
    """A [message](https://jupyter-client.readthedocs.io/en/stable/messaging.html#general-message-format)."""

    channel: Channel
    """The channel of the message."""

    header: MsgHeader
    """[ref](https://jupyter-client.readthedocs.io/en/stable/messaging.html#message-header)"""

    parent_header: MsgHeader | None
    """[ref](https://jupyter-client.readthedocs.io/en/stable/messaging.html#parent-header)"""

    metadata: dict[str, Any]
    """[ref](https://jupyter-client.readthedocs.io/en/stable/messaging.html#metadata)"""

    content: T | Content
    """[ref](https://jupyter-client.readthedocs.io/en/stable/messaging.html#metadata)

    See also:
        - [ExecuteContent][]
    """
    buffers: BuffersType
    ""

channel instance-attribute 

channel: Channel

The channel of the message.

header instance-attribute 

header: MsgHeader

ref

parent_header instance-attribute 

parent_header: MsgHeader | None

ref

metadata instance-attribute 

metadata: dict[str, Any]

ref

content instance-attribute 

content: T | Content

ref

See also

buffers instance-attribute 

buffers: BuffersType

Job

Bases: TypedDict, Generic[T]

A Message request bundle.

Returned by: Used by:

Attributes:

  • msg (Message[T]) –

    The message received over the socket.

  • ident (list[bytes]) –

    The ident associated with the message and its origin.

  • owner (Callable[[], MessageProtocol]) –

    A callable that returns the object from which a message originated.

  • received_time (float) –

    The time the message was received.

Source code in src/async_kernel/typing.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
class Job(TypedDict, Generic[T]):
    """A `Message` request bundle."""

    msg: Message[T]
    """The message received over the socket."""

    ident: list[bytes]
    """The ident associated with the message and its origin."""

    owner: Callable[[], MessageProtocol]
    """A callable that returns the object from which a message originated."""

    received_time: float
    """The time the message was received."""

msg instance-attribute 

msg: Message[T]

The message received over the socket.

ident instance-attribute 

ident: list[bytes]

The ident associated with the message and its origin.

owner instance-attribute 

owner: Callable[[], MessageProtocol]

A callable that returns the object from which a message originated.

received_time instance-attribute 

received_time: float

The time the message was received.

ExecuteContent

Bases: TypedDict

Ref.

Referenced by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
class ExecuteContent(TypedDict):
    """[Ref](https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute)."""

    code: str
    """The code to execute."""
    silent: bool
    ""
    store_history: bool
    ""
    user_expressions: dict[str, str]
    ""
    allow_stdin: bool
    ""
    stop_on_error: bool
    ""
    subshell_id: NotRequired[str | None]
    ""

code instance-attribute 

code: str

The code to execute.

silent instance-attribute 

silent: bool

store_history instance-attribute 

store_history: bool

user_expressions instance-attribute 

user_expressions: dict[str, str]

allow_stdin instance-attribute 

allow_stdin: bool

stop_on_error instance-attribute 

stop_on_error: bool

subshell_id instance-attribute 

subshell_id: NotRequired[str | None]

FixedCreate

Bases: TypedDict, Generic[S]

A TypedDict relevant to Fixed.

Referenced by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
552
553
554
555
556
557
558
class FixedCreate(TypedDict, Generic[S]):
    """A TypedDict relevant to Fixed."""

    name: str
    ""
    owner: S
    ""

name instance-attribute 

name: str

owner instance-attribute 

owner: S

FixedCreated

Bases: TypedDict, Generic[S, T]

A TypedDict relevant to Fixed.

Used by:

Attributes:

Source code in src/async_kernel/typing.py 
561
562
563
564
565
566
567
568
569
class FixedCreated(TypedDict, Generic[S, T]):
    """A TypedDict relevant to Fixed."""

    name: str
    ""
    owner: S
    ""
    obj: T
    ""

name instance-attribute 

name: str

owner instance-attribute 

owner: S

obj instance-attribute 

obj: T

RunSettings

Bases: TypedDict

A dict of settings to use with async_kernel.event_loop.run.

Subclassed by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
class RunSettings(TypedDict):
    """A dict of settings to use with [async_kernel.event_loop.run][]."""

    backend: NotRequired[Backend | Literal["asyncio", "trio"]]
    """The backend to use for the caller."""

    backend_options: NotRequired[dict | None]
    """The backend options to specify for [anyio.run][] (or `start_guest_run` when a host is specified).

    Tip:
        When there is no host and the backend is asyncio. 'loop_factory' can be specified as a function
        or importable path like `'asyncio.new_event_loop'`.
    """

    host: NotRequired[Hosts | Literal["tk", "qt"] | None]
    """The type of host where the backend will run."""

    host_options: NotRequired[dict | None]
    """Options to use when starting the host."""

backend instance-attribute 

backend: NotRequired[Backend | Literal['asyncio', 'trio']]

The backend to use for the caller.

backend_options instance-attribute 

backend_options: NotRequired[dict | None]

The backend options to specify for anyio.run (or start_guest_run when a host is specified).

Tip

When there is no host and the backend is asyncio. 'loop_factory' can be specified as a function or importable path like 'asyncio.new_event_loop'.

host instance-attribute 

host: NotRequired[Hosts | Literal['tk', 'qt'] | None]

The type of host where the backend will run.

host_options instance-attribute 

host_options: NotRequired[dict | None]

Options to use when starting the host.

CallerCreateOptions

Bases: RunSettings

Options to use when creating an instance of a Caller.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
593
594
595
596
597
598
599
600
class CallerCreateOptions(RunSettings):
    """Options to use when creating an instance of a [Caller][async_kernel.caller.Caller]."""

    name: NotRequired[str]
    """The name for the new caller instance."""

    no_debug: NotRequired[bool]
    """Disable debugpy in the thread if a new thread is created."""

name instance-attribute 

name: NotRequired[str]

The name for the new caller instance.

no_debug instance-attribute 

no_debug: NotRequired[bool]

Disable debugpy in the thread if a new thread is created.