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.

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 
48
49
50
51
52
53
54
55
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 event loop names that available to start using detail.

Referenced by: Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
58
59
60
61
62
63
64
65
66
67
68
class Hosts(enum.StrEnum):
    "An enum of event loop names that available to start using [detail][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."

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.

Channel

Bases: StrEnum

An enum of channel names Ref.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
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 
 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
156
157
158
159
160
161
162
163
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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@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 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@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 
166
167
168
169
170
171
172
173
174
175
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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

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)

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.

  • 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 
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
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).
        - 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_bool(cls, value: str | Tags, default: bool = True) -> bool:
        try:
            return value.split("=")[1].lower() == "true"
        except Exception:
            return default

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

    @classmethod
    def get_string(cls, value: str | Tags, default: str = "") -> str:
        try:
            return value.split("=")[1].strip()
        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.
    """

    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)

CallerState

Bases: Enum

The State of a async_kernel.caller.Caller.

Returned by:
Source code in src/async_kernel/typing.py 
296
297
298
299
300
301
302
303
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: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
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 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
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: list[bytearray | bytes]
    ""

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

Job

Bases: TypedDict, Generic[T]

A Message bundle.

Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
350
351
352
353
354
355
356
357
358
359
360
class Job(TypedDict, Generic[T]):
    "A `Message` bundle."

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

    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.

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 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
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.

Referenced by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
380
381
382
383
384
385
386
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 
389
390
391
392
393
394
395
396
397
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 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
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 loop is specified)."

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

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

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 loop is specified).

host instance-attribute 

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

The type of host where the backend will run.

CallerCreateOptions

Bases: RunSettings

Options to use when creating an instance of a Caller.

Referenced by: Returned by: Used by:

Attributes:

Source code in src/async_kernel/typing.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
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."

    log: NotRequired[logging.LoggerAdapter]
    "A logging adapter to use for the caller."

    "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] | None]
    "A zmq Context to use."

    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.

log instance-attribute 

log: NotRequired[LoggerAdapter]

A logging adapter to use for the caller.

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] | None]

A zmq Context to use.

no_debug instance-attribute 

no_debug: NotRequired[bool]

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