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.
get_context_hash –

Get the hash of the call stack at a certain level.

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[T_content_co] | None = None) -> Message[T_content_co] | None

Get the parent message for the current context.

Source code in src/async_kernel/utils.py

def get_parent_message(job: Job[T_content_co] | None = None, /) -> Message[T_content_co] | 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.
    """
    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]
    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) -> ErrorReply

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, /) -> t_content.ErrorReply:
    """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),
    }

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)

get_context_hash ¶

get_context_hash(level: int = 0) -> int

Get the hash of the call stack at a certain level.

This is useful to discriminate a call context in the dunder methods __enter__, __exit__, __aenter__ and __aexit__.

Parameters:

level ¶
(int, default: 0 ) –

The level up the frame stack to use starting outside the current frame.

Tips

When supporting re-entrant contexts, a list or count should be kept.
Check the hash has not already been used to prevent re-entrance.
Use atomic operations or an instance lock for thread safety, (adjust the )

Source code in src/async_kernel/utils.py

def get_context_hash(level: int = 0) -> int:
    """Get the hash of the call stack at a certain level.

    This is useful to discriminate a call context in the dunder methods `__enter__`,
     `__exit__`, `__aenter__` and `__aexit__`.

    Args:
        level: The level up the frame stack to use starting outside the current frame.

    Tips:
        - When supporting re-entrant contexts, a list or count should be kept.
        - Check the hash has not already been used to prevent re-entrance.
        - Use atomic operations or an instance lock for thread safety, (adjust
        the )
    """
    assert level >= 0
    return hash(sys._getframe(level + 2))  # pyright: ignore[reportPrivateUsage]

utils

mark_thread_pydev_do_not_trace ¶

get_kernel ¶

get_job ¶

get_parent_message ¶

get_subshell_id ¶

subshell_context ¶

`subshell_id` ¶

get_metadata ¶

get_tags ¶

get_tag_value ¶

`tag` ¶

`default` ¶

`tags` ¶

setattr_nested ¶

`obj` ¶

`name` ¶

`value` ¶

apply_settings ¶

error_to_content ¶

redirect_stdout ¶

redirect_stderr ¶

show_result ¶

get_context_hash ¶

`level` ¶

utils

mark_thread_pydev_do_not_trace ¶

get_kernel ¶

get_job ¶

get_parent_message ¶

get_subshell_id ¶

subshell_context ¶

subshell_id ¶

get_metadata ¶

get_tags ¶

get_tag_value ¶

tag ¶

default ¶

tags ¶

setattr_nested ¶

obj ¶

name ¶

value ¶

apply_settings ¶

error_to_content ¶

redirect_stdout ¶

redirect_stderr ¶

show_result ¶

get_context_hash ¶

level ¶

`subshell_id` ¶

`tag` ¶

`default` ¶

`tags` ¶

`obj` ¶

`name` ¶

`value` ¶

`level` ¶