utils

Functions:

• mark_thread_pydev_do_not_trace –

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

• get_kernel –

  Get the current kernel.

• get_job –

  Get the job for the current context.

• get_parent –

  Get the parent message for the current context.

• get_subshell_id –

  Get the subshell_id for the current context.

• subshell_context –

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

• get_metadata –

  Gets the metadata for the current context.

• get_tags –

  Gets the tags for the current context.

• get_tag_value –

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

• get_timeout –

  Gets the timeout from tags or using the current context.

• get_execution_count –

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

• setattr_nested –

  Replace an existing nested attribute/trait of an object.

• error_to_content –

  Convert the error to a dict.

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

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.

Source code in src/async_kernel/utils.py

def get_kernel() -> Kernel:
    "Get the current kernel."
    return async_kernel.Kernel()

get_job

get_job() -> Job[Any]

Get the job for the current context.

Raises:

• LookupError –

  If there is no job in the current context.

Source code in src/async_kernel/utils.py

def get_job() -> Job[Any]:
    """
    Get the job for the current context.

    Raises:
        LookupError: If there is no job in the current context.
    """
    return _job_var.get()

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

def get_parent(job: Job | None = None, /) -> Message[dict[str, Any]] | None:
    "Get the parent message for the current context."
    try:
        return (job or get_job()).get("msg")
    except LookupError:
        return None

get_subshell_id

get_subshell_id() -> str | None

Get the subshell_id for the current context.

Referenced by:

• Reference asyncshell  AsyncInteractiveSubshell

Source code in src/async_kernel/utils.py

def get_subshell_id() -> str | None:
    "Get the `subshell_id` for the current context."
    return get_kernel().shell.subshell_id

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:

• Reference asyncshell  AsyncInteractiveSubshell

Source code in src/async_kernel/utils.py

@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.
    """
    with get_kernel().subshell_manager.get_shell(subshell_id).context():
        yield

get_metadata

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

Gets the metadata for the current context.

Source code in src/async_kernel/utils.py

def get_metadata(job: Job | None = None, /) -> dict[str, Any] | None:
    "Gets the metadata for the current context."
    try:
        return (job or _job_var.get())["msg"]["metadata"]
    except Exception:
        return None

get_tags

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

Gets the tags for the current context.

Referenced by:

• Reference utils  get_tag_value

Source code in src/async_kernel/utils.py

def get_tags(job: Job | None = None, /) -> list[str]:
    "Gets the tags for the current context."
    try:
        return get_metadata(job)["tags"]  # pyright: ignore[reportOptionalSubscript]
    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

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

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

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

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

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),
    }