trulens.core.utils.python¶

trulens.core.utils.python ¶

Utilities related to core python functionalities.

Attributes¶

Thunk `module-attribute` ¶

Thunk = Callable[[], T]

A function that takes no arguments.

Classes¶

WeakWrapper `dataclass` ¶

Bases: Generic[T]

Wrap an object with a weak reference.

This is to be able to use weakref.ref on objects like lists which are otherwise not weakly referenceable. The goal of this class is to generalize weakref.ref to work with any object.

Functions¶

get ¶

get() -> T

Get the wrapped object.

OpaqueWrapper ¶

Bases: Generic[T]

Wrap an object preventing all access.

Any access except to unwrap will result in an exception with the given message.

PARAMETER	DESCRIPTION
`obj`	The object to wrap. TYPE: `T`
`e`	The exception to raise when an attribute is accessed. TYPE: `Exception`

Functions¶

unwrap ¶

unwrap() -> T

Get the wrapped object back.

SingletonPerNameMeta ¶

Bases: type

Metaclass 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¶

call ¶

__call__(*args, name: Optional[str] = None, **kwargs)

Create the singleton instance if it doesn't already exist and return it.

delete_singleton_by_name `staticmethod` ¶

delete_singleton_by_name(
    name: str, cls: Optional[Type[Any]] = 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: `str`
`cls`	The class of the singleton instance to delete. If not given, all instances with the given name are deleted. TYPE: `Optional[Type[Any]]` DEFAULT: `None`

delete_singleton `staticmethod` ¶

delete_singleton(
    obj: Type[SingletonPerNameMeta],
    name: Optional[str] = None,
)

Delete the singleton instance. Can be used for testing to create another singleton.

PydanticSingletonMeta ¶

Bases: type(BaseModel), SingletonPerNameMeta

This is the metaclass for creating Pydantic models that are also required to be singletons

Functions¶

call ¶

__call__(*args, name: Optional[str] = None, **kwargs)

Create the singleton instance if it doesn't already exist and return it.

delete_singleton_by_name `staticmethod` ¶

delete_singleton_by_name(
    name: str, cls: Optional[Type[Any]] = 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: `str`
`cls`	The class of the singleton instance to delete. If not given, all instances with the given name are deleted. TYPE: `Optional[Type[Any]]` DEFAULT: `None`

delete_singleton `staticmethod` ¶

delete_singleton(
    obj: Type[SingletonPerNameMeta],
    name: Optional[str] = None,
)

Delete the singleton instance. Can be used for testing to create another singleton.

PydanticSingleton ¶

A Pydantic model that is also a singleton by name.

This only specifies the metaclass and wraps delete_singleton_by_name.

InstanceRefMixin ¶

Mixin for classes that need to keep track of their instances.

Functions¶

get_instances `classmethod` ¶

get_instances() -> Generator[InstanceRefMixin]

Get all instances of the class.

delete_instances `classmethod` ¶

delete_instances()

Delete all instances of the class.

Functions¶

class_name ¶

class_name(obj: Union[Type, Any]) -> str

Get the class name of the given object or instance.

module_name ¶

module_name(obj: Union[ModuleType, Type, Any]) -> str

Get the module name of the given module, class, or instance.

callable_name ¶

callable_name(c: Callable)

Get the name of the given callable.

id_str ¶

id_str(obj: Any) -> str

Get the id of the given object as a string in hex.

is_really_coroutinefunction ¶

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 ¶

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.

safer_getattr ¶

safer_getattr(
    obj: Any, k: str, default: Optional[Any] = None
) -> Any

Get the attribute k of the given object.

Returns default if the attribute cannot be retrieved.

safe_getattr ¶

safe_getattr(
    obj: Any, k: str, get_prop: bool = True
) -> Any

Try to get the attribute k of the given object.

This may evaluate some code if the attribute is a property and may fail. If get_prop is False, will not return contents of properties (will raise ValueException).

safe_hasattr ¶

safe_hasattr(obj: Any, k: str) -> bool

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).

safe_issubclass ¶

safe_issubclass(cls: Type, parent: Type) -> bool

Check if the given class is a subclass of the given parent class.

code_line ¶

code_line(func, show_source: bool = False) -> Optional[str]

Get a string representation of the location of the given function func.

locals_except ¶

locals_except(*exceptions)

Get caller's locals except for the named exceptions.

for_all_methods ¶

for_all_methods(
    decorator, _except: Optional[List[str]] = None
)

Applies decorator to all methods except classmethods, private methods and the ones specified with _except.

run_before ¶

run_before(callback: Callable)

Create decorator to run the callback before the function.

superstack ¶

superstack() -> Iterator[FrameType]

Get the current stack (not including this function) with frames reaching across Tasks and threads.

caller_module_name ¶

caller_module_name(offset=0) -> str

Get the caller's (of this function) module name.

caller_module ¶

caller_module(offset=0) -> ModuleType

Get the caller's (of this function) module.

caller_frame ¶

caller_frame(offset=0) -> FrameType

Get the caller's (of this function) frame. See https://docs.python.org/3/reference/datamodel.html#frame-objects .

external_caller_frame ¶

external_caller_frame(offset=0) -> FrameType

Get the caller's (of this function) frame that is not in the trulens namespace.

RAISES	DESCRIPTION
`RuntimeError`	If no such frame is found.

caller_frameinfo ¶

caller_frameinfo(
    offset: int = 0, skip_module: Optional[str] = "trulens"
) -> Optional[FrameInfo]

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: `int` DEFAULT: `0`
`skip_module`	Skip frames from the given module. Default is "trulens". TYPE: `Optional[str]` DEFAULT: `'trulens'`

task_factory_with_stack ¶

task_factory_with_stack(
    loop, coro, *args, **kwargs
) -> Task

A task factory that annotates created tasks with stacks and context of their parents.

All of such annotated stacks can be retrieved with stack_with_tasks as one merged stack.

tru_new_event_loop ¶

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_task_stack ¶

get_task_stack(task: Task) -> Sequence[FrameType]

Get the annotated stack (if available) on the given task.

merge_stacks ¶

merge_stacks(
    s1: Iterable[FrameType], s2: Sequence[FrameType]
) -> Sequence[FrameType]

Assuming s1 is a subset of s2, combine the two stacks in presumed call order.

stack_with_tasks ¶

stack_with_tasks() -> Iterable[FrameType]

Get the current stack (not including this function) with frames reaching across Tasks.

get_all_local_in_call_stack ¶

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: `str`
`func`	Recognizer of the function to find in the call stack. TYPE: `Callable[[Callable], bool]`
`offset`	The number of top frames to skip. TYPE: `Optional[int]` DEFAULT: `1`
`skip`	A frame to skip as well. TYPE: `Optional[Any]` DEFAULT: `None`

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 `key` in the stack at all of the frames executing a function which `func` recognizes (returns True on) starting from the top of the stack except `offset` top frames. Returns None if `func` does not recognize any function in the stack.

RAISES	DESCRIPTION
`RuntimeError`	Raised 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 TP.

get_first_local_in_call_stack ¶

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.

set_context_vars_or_values ¶

set_context_vars_or_values(
    context_vars: Optional[ContextVarsOrValues] = None,
) -> Dict[ContextVar, Token]

Get the tokens for the given context variables or values.

PARAMETER	DESCRIPTION
`context_vars`	The context variables or values to get tokens for. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Dict[ContextVar, Token]`	A dictionary of context variables to tokens.

with_context ¶

with_context(
    context_vars: Optional[ContextVarsOrValues] = None,
)

Context manager to set context variables to given values.

PARAMETER	DESCRIPTION
`context_vars`	The context variables to set. If a dictionary is given, the keys are the context variables and the values are the values to set them to. If an iterable is given, it should be a list of context variables to set to their current value. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

awith_context `async` ¶

awith_context(
    context_vars: Optional[ContextVarsOrValues] = None,
)

Context manager to set context variables to given values.

PARAMETER	DESCRIPTION
`context_vars`	The context variables to set. If a dictionary is given, the keys are the context variables and the values are the values to set them to. If an iterable is given, it should be a list of context variables to set to their current value. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

wrap_awaitable ¶

wrap_awaitable(
    awaitable: Awaitable[T],
    on_await: Optional[Callable[[], Any]] = None,
    wrap: Optional[Callable[[T], T]] = None,
    on_done: Optional[Callable[[T], T]] = None,
    context_vars: Optional[ContextVarsOrValues] = None,
) -> Awaitable[T]

Wrap an awaitable in another awaitable that will call callbacks before and after the given awaitable finishes.

Important

This method captures a Context at the time this method is called and copies it over to the wrapped awaitable.

Note that the resulting awaitable needs to be awaited for the callback to eventually trigger.

PARAMETER	DESCRIPTION
`awaitable`	The awaitable to wrap. TYPE: `Awaitable[T]`
`on_await`	The callback to call when the wrapper awaitable is awaited but before the wrapped awaitable is awaited. TYPE: `Optional[Callable[[], Any]]` DEFAULT: `None`
`wrap`	The callback to call with the result of the wrapped awaitable once it is ready. This should return the value or a wrapped version. TYPE: `Optional[Callable[[T], T]]` DEFAULT: `None`
`on_done`	For compatibility with generators, this is called after wrap. TYPE: `Optional[Callable[[T], T]]` DEFAULT: `None`
`context_vars`	The context variables to copy over to the wrapped awaitable. If None, all context variables are copied. See with_context. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

wrap_generator ¶

wrap_generator(
    gen: Generator[T, None, None],
    on_iter: Optional[Callable[[], Any]] = None,
    wrap: Optional[Callable[[T], T]] = None,
    on_done: Optional[Callable[[List[T]], Any]] = None,
    context_vars: Optional[ContextVarsOrValues] = 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: `Generator[T, None, None]`
`on_iter`	The callback to call when the wrapper generator is created but before a first iteration is produced. TYPE: `Optional[Callable[[], Any]]` DEFAULT: `None`
`wrap`	The callback to call with the result of each iteration of the wrapped generator. This should return the value or a wrapped version. TYPE: `Optional[Callable[[T], T]]` DEFAULT: `None`
`on_done`	The callback to call when the wrapped generator is exhausted. TYPE: `Optional[Callable[[List[T]], Any]]` DEFAULT: `None`
`context_vars`	The context variables to copy over to the wrapped generator. If None, all context variables are taken with their present values. See with_context. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

wrap_async_generator ¶

wrap_async_generator(
    gen: AsyncGenerator[T, None],
    on_iter: Optional[Callable[[], Any]] = None,
    wrap: Optional[Callable[[T], T]] = None,
    on_done: Optional[Callable[[List[T]], Any]] = None,
    context_vars: Optional[ContextVarsOrValues] = None,
) -> AsyncGenerator[T, 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: `AsyncGenerator[T, None]`
`on_iter`	The callback to call when the wrapper generator is created but before a first iteration is produced. TYPE: `Optional[Callable[[], Any]]` DEFAULT: `None`
`wrap`	The callback to call with the result of each iteration of the wrapped generator. TYPE: `Optional[Callable[[T], T]]` DEFAULT: `None`
`on_done`	The callback to call when the wrapped generator is exhausted. TYPE: `Optional[Callable[[List[T]], Any]]` DEFAULT: `None`
`context_vars`	The context variables to copy over to the wrapped generator. If None, all context variables are taken with their present values. See with_context. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

is_lazy ¶

is_lazy(obj)

Check if the given object is lazy.

An object is considered lazy if it is a generator or an awaitable.

wrap_lazy ¶

wrap_lazy(
    obj: Any,
    on_start: Optional[Callable[[], None]] = None,
    wrap: Optional[Callable[[T], T]] = None,
    on_done: Optional[Callable[[Any], Any]] = None,
    context_vars: Optional[ContextVarsOrValues] = None,
) -> Any

Wrap a lazy value in one that will call callbacks at various points in the generation process.

PARAMETER	DESCRIPTION
`obj`	The lazy value. TYPE: `Any`
`on_start`	The callback to call when the wrapper is created. TYPE: `Optional[Callable[[], None]]` DEFAULT: `None`
`wrap`	The callback to call with the result of each iteration of the wrapped generator or the result of an awaitable. This should return the value or a wrapped version. TYPE: `Optional[Callable[[T], T]]` DEFAULT: `None`
`on_done`	The callback to call when the wrapped generator is exhausted or awaitable is ready. TYPE: `Optional[Callable[[Any], Any]]` DEFAULT: `None`
`context_vars`	The context variables to copy over to the wrapped generator. If None, all context variables are taken with their present values. See with_context. TYPE: `Optional[ContextVarsOrValues]` DEFAULT: `None`

wrap_until_eager ¶

wrap_until_eager(
    obj,
    on_eager: Optional[Callable[[Any], T]] = None,
    context_vars: Optional[ContextVarsOrValues] = None,
) -> T | Sequence[T]

Wrap a lazy value in one that will call callbacks one the final non-lazy values.

Arts

obj: The lazy value.

on_eager: The callback to call with the final value of the wrapped generator or the result of an awaitable. This should return the value or a wrapped version.

context_vars: The context variables to copy over to the wrapped generator. If None, all context variables are taken with their present values. See with_context.

context_id ¶

context_id() -> str

Return a short representation of context that includes the thread and async task identifiers.

trulens.core.utils.python¶