trulens.core.utils.python¶
Utilities related to core python functionalities.
Attributes¶
module-attribute
¶
NoneType = NoneType
Alias for types.NoneType .
In python < 3.10, it is defined as type(None) instead.
Classes¶
Alias for concurrent.futures.Future.
In python < 3.9, a sublcass of concurrent.futures.Future with
Generic[A] is used instead.
Alias for queue.Queue .
In python < 3.9, a sublcass of queue.Queue with
Generic[A] is used instead.
dataclass
¶
Bases: Generic[T]
Information about a singleton instance.
Attributes¶
instance-attribute
¶
frame: Any
The frame where the singleton was created.
This is used for showing "already created" warnings.
class-attribute
instance-attribute
¶
The name of the singleton instance.
This is used for the SingletonPerName mechanism to have a separate singleton for each unique name (and class).
Functions¶
Class for creating singleton instances except there being one instance max,
there is one max per different name argument. If name is never given,
reverts to normal singleton behavior.
Functions¶
__new__(
*args, name: Optional[str] = None, **kwargs
) -> SingletonPerName
Create the singleton instance if it doesn't already exist and return it.
staticmethod
¶
delete_singleton_by_name(
name: str, cls: Optional[Type[SingletonPerName]] = None
)
Delete the singleton instance with the given name.
This can be used for testing to create another singleton.
| PARAMETER | DESCRIPTION |
|---|---|
name |
The name of the singleton instance to delete.
TYPE:
|
cls |
The class of the singleton instance to delete. If not given, all instances with the given name are deleted.
TYPE:
|
delete_singleton()
Delete the singleton instance. Can be used for testing to create another singleton.
Functions¶
getmembers_static(obj, predicate=None)
Implementation of inspect.getmembers_static for python < 3.11.
Get the class name of the given object or instance.
module_name(obj: Union[ModuleType, Type, Any]) -> str
Get the module name of the given module, class, or instance.
is_really_coroutinefunction(func) -> bool
Determine whether the given function is a coroutine function.
Warning
Inspect checkers for async functions do not work on openai clients,
perhaps because they use @typing.overload. Because of that, we detect
them by checking __wrapped__ attribute instead. Note that the inspect
docs suggest they should be able to handle wrapped functions but perhaps
they handle different type of wrapping? See
https://docs.python.org/3/library/inspect.html#inspect.iscoroutinefunction
. Another place they do not work is the decorator langchain uses to mark
deprecated functions.
safe_signature(func_or_obj: Any)
Get the signature of the given function.
Sometimes signature fails for wrapped callables and in those cases we check
for __call__ attribute and use that instead.
Check if the given object has the given attribute.
Attempts to use static checks (see inspect.getattr_static) to avoid any side effects of attribute access (i.e. for properties).
Check if the given class is a subclass of the given parent class.
Get a string representation of the location of the given function
func.
Applies decorator to all methods except classmethods, private methods and
the ones specified with _except.
run_before(callback: Callable)
Create decorator to run the callback before the function.
Get the current stack (not including this function) with frames reaching across Tasks and threads.
caller_module_name(offset=0) -> str
Get the caller's (of this function) module name.
caller_frame(offset=0) -> FrameType
Get the caller's (of this function) frame. See https://docs.python.org/3/reference/datamodel.html#frame-objects .
Get the caller's (of this function) frameinfo. See https://docs.python.org/3/reference/datamodel.html#frame-objects .
| PARAMETER | DESCRIPTION |
|---|---|
offset |
The number of frames to skip. Default is 0.
TYPE:
|
skip_module |
Skip frames from the given module. Default is "trulens". |
A task factory that annotates created tasks with stacks of their parents.
All of such annotated stacks can be retrieved with stack_with_tasks as one merged stack.
tru_new_event_loop()
Replacement for new_event_loop that sets the task factory to make tasks that copy the stack from their creators.
Get the annotated stack (if available) on the given task.
Assuming s1 is a subset of s2, combine the two stacks in presumed call
order.
Get the current stack (not including this function) with frames reaching across Tasks.
get_all_local_in_call_stack(
key: str,
func: Callable[[Callable], bool],
offset: Optional[int] = 1,
skip: Optional[Any] = None,
) -> Iterator[Any]
Find locals in call stack by name.
| PARAMETER | DESCRIPTION |
|---|---|
key |
The name of the local variable to look for.
TYPE:
|
func |
Recognizer of the function to find in the call stack. |
offset |
The number of top frames to skip. |
skip |
A frame to skip as well. |
Note
offset is unreliable for skipping the intended frame when operating
with async tasks. In those cases, the skip argument is more reliable.
| RETURNS | DESCRIPTION |
|---|---|
Iterator[Any]
|
An iterator over the values of the local variable named Returns None if |
| RAISES | DESCRIPTION |
|---|---|
RuntimeError
|
Raised if a function is recognized but does not have |
This method works across threads as long as they are started using TP.
get_first_local_in_call_stack(
key: str,
func: Callable[[Callable], bool],
offset: Optional[int] = 1,
skip: Optional[Any] = None,
) -> Optional[Any]
Get the value of the local variable named key in the stack at the nearest
frame executing a function which func recognizes (returns True on)
starting from the top of the stack except offset top frames. If skip
frame is provided, it is skipped as well. Returns None if func does not
recognize the correct function. Raises RuntimeError if a function is
recognized but does not have key in its locals.
This method works across threads as long as they are started using the TP class above.
NOTE: offset is unreliable for skipping the intended frame when operating
with async tasks. In those cases, the skip argument is more reliable.
wrap_awaitable(
awaitable: Awaitable[T],
on_await: Optional[Callable[[], Any]] = None,
on_done: Optional[Callable[[T], Any]] = None,
) -> Awaitable[T]
Wrap an awaitable in another awaitable that will call callbacks before and after the given awaitable finishes.
Note that the resulting awaitable needs to be awaited for the callback to eventually trigger.
| PARAMETER | DESCRIPTION |
|---|---|
awaitable |
The awaitable to wrap.
TYPE:
|
on_await |
The callback to call when the wrapper awaitable is awaited but before the wrapped awaitable is awaited. |
on_done |
The callback to call with the result of the wrapped awaitable once it is ready. |
wrap_generator(
gen: Generator[T, None, None],
on_iter: Optional[Callable[[], Any]] = None,
on_next: Optional[Callable[[T], Any]] = None,
on_done: Optional[Callable[[], Any]] = None,
) -> Generator[T, None, None]
Wrap a generator in another generator that will call callbacks at various points in the generation process.
| PARAMETER | DESCRIPTION |
|---|---|
gen |
The generator to wrap.
TYPE:
|
on_iter |
The callback to call when the wrapper generator is created but before a first iteration is produced. |
on_next |
The callback to call with the result of each iteration of the wrapped generator. |
on_done |
The callback to call when the wrapped generator is exhausted. |