Skip to content

typing

Classes:

Attributes:

  • NoValue –

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

Returned by: Used by:

HandlerType module-attribute 

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

A TypeAlias for the handler of message requests.

Used by:

SocketID

Bases: StrEnum

Mapping of Kernel.port_<id> for sockets. Ref.

Returned by:

Attributes:

Source code in src/async_kernel/typing.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
class SocketID(enum.StrEnum):
    "Mapping of `Kernel.port_<id>` for sockets. [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 Messages.

receive_msg_loop uses get_run_mode to map the message type and channel (shell or control) to the RunMode.

Cell overrides

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

Top line comment: 

# task
or

```python
##task
```
Returned by:

Attributes:

Source code in src/async_kernel/typing.py 
 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
class RunMode(enum.StrEnum):
    """
    An Enum of the run modes available for handling [Messages][async_kernel.typing.Message].

    [receive_msg_loop][async_kernel.kernel.Kernel.receive_msg_loop] uses [get_run_mode][async_kernel.kernel.Kernel.get_run_mode]
    to map the message type and channel (`shell` or `control`) to the `RunMode`.

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

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

            ```python
            ##task
            ```
    """

    @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)

    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][] to start use a 'worker'."

    direct = "direct"
    """
    Run the message handler using [async_kernel.caller.Caller.call_direct][].

    Warning: 
        - This mode runs directly in the caller scheduler as soon as it is received.
        - Use this only for fast running high priority code.
    """

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 start use a 'worker'.

direct class-attribute instance-attribute 

direct = 'direct'

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

Warning
  • This mode runs directly in the caller scheduler as soon as it is received.
  • Use this only for fast running high priority code.

MsgType

Bases: StrEnum

An enumeration of Message msg_type for shell and control messages.

Some message types are on the control channel only.

Referenced by: Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
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
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)"

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

comm_msg class-attribute instance-attribute 

comm_msg = 'comm_msg'

async_kernel.kernel.Kernel.comm_msg

comm_close class-attribute instance-attribute 

comm_close = 'comm_close'

async_kernel.kernel.Kernel.comm_close

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)

Tags

Bases: StrEnum

Tags recognised by the shell.

Info

Tags are can be added per cell.

Used by:

Attributes:

  • raises_exception –

    Indicates the cell should expect an exception to be raised.

  • suppress_error –

    Suppress exceptions that occur during execution of the code cell.

  • 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 
181
182
183
184
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
class Tags(enum.StrEnum):
    """
    Tags recognised by the [shell][async_kernel.asyncshell.AsyncInteractiveShell].

    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).
        - VScode: 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)

    def get_bool(self, value: str | Tags, default: bool = True) -> bool:
        try:
            return value.split("=")[1].lower() == "true"
        except Exception:
            return default

    def get_float(self, value: str | Tags, default: float = 0.0) -> float:
        try:
            return float(value.split("=")[1])
        except Exception:
            return default

    def get_string(self, value: str | Tags, default: str = "") -> str:
        try:
            return value.split("=")[1]
        except Exception:
            return default

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

    suppress_error = "suppress-error"
    """
    Suppress exceptions that occur during execution of the code cell.

    The default message is '⚠'

    Examples:

        - suppress-error 
        - suppress-error=The suppression message.

    Warning:
        The code block will return as 'ok' (not published).
    """

    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.

suppress_error class-attribute instance-attribute 

suppress_error = 'suppress-error'

Suppress exceptions that occur during execution of the code cell.

The default message is '⚠'

Examples:

- suppress-error 
- suppress-error=The suppression message.
Warning

The code block will return as 'ok' (not published).

Referenced by:

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)

CallerState

Bases: Enum

The State of a async_kernel.caller.Caller.

Returned by:
Source code in src/async_kernel/typing.py 
275
276
277
278
279
280
281
282
class CallerState(enum.Enum):
    "The State of a [async_kernel.caller.Caller][]."

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

MsgHeader

Bases: TypedDict

A message header.

Returned by:

Attributes:

Source code in src/async_kernel/typing.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
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
    ""
    msg_type: MsgType
    ""
    version: str
    ""

msg_id instance-attribute 

msg_id: str

session instance-attribute 

session: str

username instance-attribute 

username: str

date instance-attribute 

date: str

msg_type instance-attribute 

msg_type: MsgType

version instance-attribute 

version: str

Message

Bases: TypedDict, Generic[T]

A message.

Referenced by: Returned by:

Attributes:

Source code in src/async_kernel/typing.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
class Message(TypedDict, Generic[T]):
    "A [message](https://jupyter-client.readthedocs.io/en/stable/messaging.html#general-message-format)."

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

    parent_header: MsgHeader
    "[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: list[bytearray | bytes]
    ""

header instance-attribute 

header: MsgHeader

ref

parent_header instance-attribute 

parent_header: MsgHeader

ref

metadata instance-attribute 

metadata: dict[str, Any]

ref

content instance-attribute 

content: T | Content

ref

See also

buffers instance-attribute 

buffers: list[bytearray | bytes]

Job

Bases: TypedDict, Generic[T]

A Message bundle.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
324
325
326
327
328
329
330
331
332
333
334
class Job(TypedDict, Generic[T]):
    "A `Message` bundle."

    msg: Message[T]
    "The message received over the socket."
    socket_id: Literal[SocketID.control, SocketID.shell]
    "The channel over which the socket was received."
    ident: bytes | list[bytes]
    "The ident associated with the message and its origin."
    received_time: float
    "The time the message was received."

msg instance-attribute 

msg: Message[T]

The message received over the socket.

socket_id instance-attribute 

socket_id: Literal[control, shell]

The channel over which the socket was received.

ident instance-attribute 

ident: bytes | list[bytes]

The ident associated with the message and its origin.

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 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
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
    ""

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

FixedCreate

Bases: TypedDict, Generic[S]

A TypedDict relevant to Fixed.

Source code in src/async_kernel/typing.py 
354
355
356
357
358
class FixedCreate(TypedDict, Generic[S]):
    "A TypedDict relevant to Fixed."

    name: str
    owner: S

FixedCreated

Bases: TypedDict, Generic[S, T]

A TypedDict relevant to Fixed.

Source code in src/async_kernel/typing.py 
361
362
363
364
365
366
class FixedCreated(TypedDict, Generic[S, T]):
    "A TypedDict relevant to Fixed."

    name: str
    owner: S
    obj: T

PendingCreateOptions

Bases: TypedDict

Options to pass when creating a new Pending.

Used by:

Attributes:

Source code in src/async_kernel/typing.py 
369
370
371
372
373
class PendingCreateOptions(TypedDict):
    "Options to pass when creating a new [Pending][async_kernel.pending.Pending]."

    allow_tracking: NotRequired[bool]
    "Add the pending to all [pending trackers][async_kernel.pending.PendingTracker] (default=`True`)."

allow_tracking instance-attribute 

allow_tracking: NotRequired[bool]

Add the pending to all pending trackers (default=True).

Referenced by:

CallerCreateOptions

Bases: TypedDict

Options to use when creating an instance of a Caller.

Used by:

Attributes:

Source code in src/async_kernel/typing.py 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
class CallerCreateOptions(TypedDict):
    "Options to use when creating an instance of a [Caller][async_kernel.caller.Caller]."

    name: NotRequired[str]
    "The name for the new caller instance."
    log: NotRequired[logging.LoggerAdapter]
    "A logging adapter to use for the caller."
    backend: NotRequired[Backend | Literal["trio", "asyncio"]]
    "The backend to specify when calling [anyio.run][]."
    backend_options: NotRequired[dict | None]
    "Options to pass when calling [anyio.run][]."
    protected: NotRequired[bool]
    "The caller should be protected against accidental closure (default=`False`)."
    zmq_context: NotRequired[zmq.Context[Any]]
    "A zmq Context to use."

name instance-attribute 

name: NotRequired[str]

The name for the new caller instance.

log instance-attribute 

log: NotRequired[LoggerAdapter]

A logging adapter to use for the caller.

backend instance-attribute 

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

The backend to specify when calling anyio.run.

backend_options instance-attribute 

backend_options: NotRequired[dict | None]

Options to pass when calling anyio.run.

protected instance-attribute 

protected: NotRequired[bool]

The caller should be protected against accidental closure (default=False).

zmq_context instance-attribute 

zmq_context: NotRequired[Context[Any]]

A zmq Context to use.