orangeqs.juice._util#

Internal utilities for Juice.

Module Contents#

Functions#

warn_deprecated_module

Warn that old is a deprecated alias, kept for backwards compatibility.

add_note

Add a note to an exception, backported for Python <3.11.

rerun_on_exception

Rerun async function on exception.

spawn_thread

Spawn a thread to run an async function.

load_object_reference

Load a Python object from an importable reference.

API#

orangeqs.juice._util.warn_deprecated_module(old: str, new: str) → None#

Warn that old is a deprecated alias, kept for backwards compatibility.

Emits both a DeprecationWarning (for callers with warnings enabled) and a logger warning (since a long-running service may not surface warnings in its logs otherwise). Intended to be called once at the top of a deprecated shim module.

Parameters#

  • old (str): The deprecated module’s dotted path.

  • new (str): The dotted path of the module that replaces it.

orangeqs.juice._util.add_note(exception: orangeqs.juice._util._BaseException, note: str) → orangeqs.juice._util._BaseException#

Add a note to an exception, backported for Python <3.11.

Ensures the exception is rendered the same way as BaseException.add_note() which was added in Python 3.11. See https://docs.python.org/3/library/exceptions.html#BaseException.add_note.

Parameters#

  • exception (BaseException): The exception to attach the note to.

  • note (str): The note to attach.

orangeqs.juice._util.rerun_on_exception(coro: collections.abc.Callable[orangeqs.juice._util._P, collections.abc.Awaitable[orangeqs.juice._util._T]] | None = None, *, sleeptime_s: float = 5.0) → collections.abc.Callable[[collections.abc.Callable[orangeqs.juice._util._P, collections.abc.Awaitable[orangeqs.juice._util._T]]], collections.abc.Callable[orangeqs.juice._util._P, collections.abc.Awaitable[orangeqs.juice._util._T]]] | collections.abc.Callable[orangeqs.juice._util._P, collections.abc.Awaitable[orangeqs.juice._util._T]]#

Rerun async function on exception.

To prevent busy threads, sleep before rerun. The sleeptime_s parameter specifies the time to sleep before retrying. This decorator can both be used with and without parameters: @rerun_on_exception or @rerun_on_exception(sleeptime_s=10).

Parameters#

  • sleeptime_s (float, optional): Time to sleep before retrying, in seconds. Default is 5.0 seconds.

orangeqs.juice._util.spawn_thread(future: collections.abc.Awaitable[None]) → threading.Thread#

Spawn a thread to run an async function.

orangeqs.juice._util.load_object_reference(object_reference: str) → Any#

Load a Python object from an importable reference.

Can be used to load a entry point object reference.

Parameters#

  • object_reference (str): The reference points to a Python object. It is either in the form importable.module, or importable.module:object.attr. Each of the parts delimited by dots and the colon is a valid Python identifier.

Returns#

  • (Any): The Python object pointed to by the reference.