utils

Provides publicly utility functions.

Used by:

• Reference interface
  •  Interface  force_shutdown_delay
  •  Interface  force_shutdown_delay

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_message –

  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.

• setattr_nested –

  Replace an existing nested attribute/trait of an object.

• apply_settings –

  Apply the settings onto the object.

• error_to_content –

  Convert the error to a dict.

• redirect_stdout –

  Re-direct sys.stdout generated in the current context.

• redirect_stderr –

  Re-direct sys.stderr generated in the current context.

• show_result –

  A context where show_result is enabled/disabled.

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]

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."""
    import async_kernel.interface  # noqa: PLC0415

    if not async_kernel.interface.Interface.initialized():
        msg = "A kernel interface is not started!"
        raise RuntimeError(msg)
    return async_kernel.interface.Interface.instance().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_message

get_parent_message(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_message(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.

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.

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().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: Iterable[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

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

  A list of tags to search. When not provided 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: Iterable[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 provided [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:
            return tag.get_value(t, default)
    return default

setattr_nested

setattr_nested(
    obj: object, name: str, value: str | Any, *, _return_value=False
) -> 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, *, _return_value=False) -> 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.
    """
    try:
        if len(bits := name.split(".")) > 1:
            try:
                obj = getattr(obj, bits[0])
            except Exception:
                return {}
            value = setattr_nested(obj, ".".join(bits[1:]), value, _return_value=True)
            return value if _return_value else {name: 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 value if _return_value else {name: value}  # pyright: ignore[reportReturnType]
    except Exception:
        if not _return_value:
            raise
    return {}

apply_settings

apply_settings(obj: object, settings: Mapping[str, Any]) -> dict[str, Any]

Apply the settings onto the object.

Returns:

• dict ( dict[str, Any] ) –

  A copy of the settings that were applied.

Notes

• If flags are included using the pattern 'flags': iterable(str), the flags are interpreted as boolean values. Generally, it is preferred to specify the value in the settings explicitly.

Source code in src/async_kernel/utils.py

def apply_settings(obj: object, settings: Mapping[str, Any]) -> dict[str, Any]:
    """Apply the settings onto the object.

    Returns:
        dict: A copy of the settings that were applied.

    Notes:
        - If flags are included using the pattern `'flags': iterable(str)`, the flags
            are interpreted as boolean values. Generally, it is preferred to specify
            the value in the settings explicitly.
    """
    values = {}
    for k, v in settings.items():
        if k == "flags":
            for flag in (f.strip("-") for f in v):
                values.update(setattr_nested(obj, flag.strip("no-"), not flag.startswith("no-")))
        else:
            values.update(setattr_nested(obj, k, v))
    return values

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": MsgType.iopub_error,
        "ename": type(error).__name__,
        "evalue": str(error),
        "traceback": traceback.format_exception(error),
    }

redirect_stdout

redirect_stdout(stream: _SupportsRedirectT) -> Generator[_SupportsRedirectT, Any, None]

Re-direct sys.stdout generated in the current context.

See Also

• contextlib.redirect_stdout

Referenced by:

• Reference ipshell  IPShell  system

Source code in src/async_kernel/utils.py

@contextmanager
def redirect_stdout(stream: _SupportsRedirectT, /) -> Generator[_SupportsRedirectT, Any, None]:
    """Re-direct [sys.stdout][] generated in the current context.

    See Also:
        - [contextlib.redirect_stdout][]
    """
    token = _stdout_context.set(stream)
    try:
        yield stream
    finally:
        _stdout_context.reset(token)

redirect_stderr

redirect_stderr(stream: _SupportsRedirectT) -> Generator[_SupportsRedirectT, Any, None]

Re-direct sys.stderr generated in the current context.

See Also

• contextlib.redirect_stderr

Referenced by:

• Reference ipshell  IPShell  system

Source code in src/async_kernel/utils.py

@contextmanager
def redirect_stderr(stream: _SupportsRedirectT, /) -> Generator[_SupportsRedirectT, Any, None]:
    """Re-direct [sys.stderr][] generated in the current context.

    See Also:
        - [contextlib.redirect_stderr][]
    """
    token = _stderr_context.set(stream)
    try:
        yield stream
    finally:
        _stderr_context.reset(token)

show_result

show_result(show: bool = True) -> Generator[None, Any, None]

A context where show_result is enabled/disabled.

Source code in src/async_kernel/utils.py

@contextmanager
def show_result(show: bool = True, /) -> Generator[None, Any, None]:
    """A context where `show_result` is enabled/disabled."""
    token = _show_result_context.set(show)
    try:
        yield
    finally:
        if token:
            _show_result_context.reset(token)