Skip to content

utils

Functions:

mark_thread_pydev_do_not_trace 

mark_thread_pydev_do_not_trace(thread: Thread | None = None, *, remove=False) -> None

Modifies the given thread's attributes to hide or unhide it from the debugger (e.g., debugpy).

Source code in src/async_kernel/utils.py 
46
47
48
49
50
51
def mark_thread_pydev_do_not_trace(thread: threading.Thread | None = None, *, remove=False) -> None:
    """Modifies the given thread's attributes to hide or unhide it from the debugger (e.g., debugpy)."""
    thread = thread or threading.current_thread()
    thread.pydev_do_not_trace = not remove  # pyright: ignore[reportAttributeAccessIssue]
    thread.is_pydev_daemon_thread = not remove  # pyright: ignore[reportAttributeAccessIssue]
    return

get_kernel 

get_kernel() -> Kernel

Get the current kernel.

Referenced by:
Source code in src/async_kernel/utils.py 
54
55
56
def get_kernel() -> Kernel:
    "Get the current kernel."
    return async_kernel.Kernel()

get_job 

get_job() -> Job[dict] | dict

Get the job for the current context.

Source code in src/async_kernel/utils.py 
59
60
61
62
63
64
def get_job() -> Job[dict] | dict:
    "Get the job for the current context."
    try:
        return _job_var.get()
    except Exception:
        return {}

get_parent 

get_parent(job: Job | None = None) -> Message[dict[str, Any]] | None

Get the parent message for the current context.

Source code in src/async_kernel/utils.py 
67
68
69
def get_parent(job: Job | None = None, /) -> Message[dict[str, Any]] | None:
    "Get the [parent message]() for the current context."
    return (job or get_job()).get("msg")

get_subshell_id 

get_subshell_id() -> str | None

Get the subshell_id for the current context.

Referenced by:
Source code in src/async_kernel/utils.py 
72
73
74
def get_subshell_id() -> str | None:
    "Get the `subshell_id` for the current context."
    return SubshellPendingManager._contextvar.get()  # pyright: ignore[reportPrivateUsage]

subshell_context 

subshell_context(subshell_id: str | None) -> Generator[None, Any, None]

A context manager to work in the context of a shell or subshell.

Parameters:

  • subshell_id

    (str | None) –

    An existing subshell or the main shell if subshell_id is None.

Referenced by:
Source code in src/async_kernel/utils.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
@contextmanager
def subshell_context(subshell_id: str | None) -> Generator[None, Any, None]:
    """A context manager to work in the context of a shell or subshell.

    Args:
        subshell_id: An existing subshell or the main shell if subshell_id is None.
    """
    shell = get_kernel().subshell_manager.get_shell(subshell_id)  # use the shell for validation.
    token = SubshellPendingManager._contextvar.set(shell.subshell_id)  # pyright: ignore[reportPrivateUsage]
    try:
        yield
    finally:
        SubshellPendingManager._contextvar.reset(token)  # pyright: ignore[reportPrivateUsage]

get_metadata 

get_metadata(job: Job | None = None) -> dict[str, Any]

Gets the metadata for the current context.

Source code in src/async_kernel/utils.py 
92
93
94
95
96
97
def get_metadata(job: Job | None = None, /) -> dict[str, Any]:
    "Gets the metadata for the current context."
    try:
        return (job or _job_var.get())["msg"]["metadata"]
    except Exception:
        return {}

get_tags 

get_tags(job: Job | None = None) -> list[str]

Gets the tags for the current context.

Referenced by:
Source code in src/async_kernel/utils.py 
100
101
102
103
104
105
def get_tags(job: Job | None = None, /) -> list[str]:
    "Gets the tags for the current context."
    try:
        return get_metadata(job)["tags"]
    except Exception:
        return []

get_tag_value 

get_tag_value(
    tag: Tags, default: _TagType, /, *, tags: list[str] | None = None
) -> _TagType

Get the value for the tag from a collection of tags.

Parameters:

  • tag

    (Tags) –

    The tag to get the value from.

  • default

    (_TagType) –

    The default value if a tag is not found. The default is also used to determine the type for conversion of the value.

  • tags

    (list[str] | None, default: None ) –

    A list of tags to search. When not provide get_tags is used.

The tag value is the value trailing behind =. The value is transformed according to the type of the default.

Source code in src/async_kernel/utils.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_tag_value(tag: Tags, default: _TagType, /, *, tags: list[str] | None = None) -> _TagType:
    """
    Get the value for the tag from a collection of tags.

    Args:
        tag: The tag to get the value from.
        default: The default value if a tag is not found. The default is also used to determine the type for conversion of the value.
        tags: A list of tags to search. When not provide [get_tags][] is used.

    The tag value is the value trailing behind <tag>=<value>. The value is transformed according to
    the type of the default.
    """
    for t in tags if tags is not None else get_tags():
        if t == tag:
            if isinstance(default, float):
                return tag.get_float(t, default)
            if isinstance(default, bool):
                return tag.get_bool(t, default)
            if isinstance(default, str):
                return tag.get_string(t, default)
            return int(tag.get_float(t, default))
    return default

get_timeout 

get_timeout(*, tags: list[str] | None = None) -> float

Gets the timeout from tags or using the current context.

Source code in src/async_kernel/utils.py 
135
136
137
138
139
def get_timeout(*, tags: list[str] | None = None) -> float:
    "Gets the timeout from tags or using the current context."
    if math.isnan(timeout := get_tag_value(Tags.timeout, math.nan, tags=tags)):
        return get_kernel().shell.timeout
    return max(timeout, 0.0)

get_execution_count 

get_execution_count() -> int

Gets the execution count for the current context, defaults to the current kernel count.

Source code in src/async_kernel/utils.py 
142
143
144
145
def get_execution_count() -> int:
    "Gets the execution count for the current context, defaults to the current kernel count."

    return get_kernel().shell.execution_count

setattr_nested 

setattr_nested(obj: object, name: str, value: str | Any) -> dict[str, Any]

Replace an existing nested attribute/trait of an object.

If the attribute name contains dots, it is interpreted as a nested attribute. For example, if name is "a.b.c", then the code will attempt to set obj.a.b.c to value.

Parameters:

  • obj

    (object) –

    The object to set the attribute on.

  • name

    (str) –

    The name of the attribute to set.

  • value

    (str | Any) –

    The value to set the attribute to.

Returns:

  • dict[str, Any] –

    The mapping of the name to the set value if the value has been set.

  • dict[str, Any] –

    An empty dict indicates the value was not set.

Source code in src/async_kernel/utils.py 
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
def setattr_nested(obj: object, name: str, value: str | Any) -> dict[str, Any]:
    """
    Replace an existing nested attribute/trait of an object.

    If the attribute name contains dots, it is interpreted as a nested attribute.
    For example, if name is "a.b.c", then the code will attempt to set obj.a.b.c to value.

    Args:
        obj: The object to set the attribute on.
        name: The name of the attribute to set.
        value: The value to set the attribute to.

    Returns:
        The mapping of the name to the set value if the value has been set.
        An empty dict indicates the value was not set.
    """
    import traitlets  # noqa: PLC0415

    if len(bits := name.split(".")) > 1:
        try:
            obj = getattr(obj, bits[0])
        except Exception:
            return {}
        setattr_nested(obj, ".".join(bits[1:]), value)
    if (isinstance(obj, traitlets.HasTraits) and obj.has_trait(name)) or hasattr(obj, name):
        try:
            setattr(obj, name, value)
        except Exception:
            setattr(obj, name, eval(value))
        return {name: getattr(obj, name)}
    return {}

error_to_content 

error_to_content(error: BaseException) -> Content

Convert the error to a dict.

ref: https://jupyter-client.readthedocs.io/en/stable/messaging.html#request-reply

Source code in src/async_kernel/utils.py 
181
182
183
184
185
186
187
188
189
190
191
192
def error_to_content(error: BaseException, /) -> Content:
    """
    Convert the error to a dict.

    ref: https://jupyter-client.readthedocs.io/en/stable/messaging.html#request-reply
    """
    return {
        "status": "error",
        "ename": type(error).__name__,
        "evalue": str(error),
        "traceback": traceback.format_exception(error),
    }