Python Utilities¶
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.
instance-attribute
¶
The name of the singleton instance.
This is used for the SingletonPerName mechanism to have a seperate singleton for each unique name (and class).
Functions¶
Bases: Generic[T]
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 behaviour.
Functions¶
staticmethod
¶
delete_singleton_by_name(name: str, cls: 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¶
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.
caller_frame(offset=0) -> 'frame'
Get the caller's (of this function) frame. See https://docs.python.org/3/reference/datamodel.html#frame-objects .
caller_frameinfo(offset: int = 0, skip_module: Optional[str] = 'trulens_eval') -> 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:
|
skip_module |
Skip frames from the given module. Default is "trulens_eval". |
task_factory_with_stack(loop, coro, *args, **kwargs) -> Sequence['frame']
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.
stack_with_tasks() -> Sequence['frame']
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. |
Serialization of Python objects¶
In order to serialize (and optionally deserialize) python entities while still being able to inspect them in their serialized form, we employ several storage classes that mimic basic python entities:
| Serializable representation | Python entity |
|---|---|
| Class | (python) class |
| Module | (python) module |
| Obj | (python) object |
| Function | (python) function |
| Method | (python) method |
Classes¶
Bases: SerialModel
A python class. Should be enough to deserialize the constructor. Also includes bases so that we can query subtyping relationships without deserializing the class first.
Bases: SerialModel
An object that may or may not be loadable from its serialized form. Do not
use for base types that don't have a class. Loadable if init_bindings is
not None.
Bases: SerialModel
Bases: FunctionOrMethod
A python method. A method belongs to some class in some module and must have
a pre-bound self object. The location of the method is encoded in obj
alongside self. If obj is Obj with init_bindings, this method should be
deserializable.
Bases: FunctionOrMethod
A python function. Could be a static method inside a class (not instance of the class).
Bases: BaseModel
Mixin to track class information to aid in querying serialized components without having to load them.
Functions¶
staticmethod
¶
load(obj, *args, **kwargs)
Deserialize/load this object using the class information in tru_class_info to lookup the actual class that will do the deserialization.
classmethod
¶
model_validate(*args, **kwargs) -> Any
Deserialized a jsonized version of the app into the instance of the class it was serialized from.
Note
This process uses extra information stored in the jsonized object and handled by WithClassInfo.
Functions¶
is_noserio(obj)
Determines whether the given json object represents some non-serializable
object. See noserio.
Create a json structure to represent a non-serializable object. Any additional keyword arguments are included.
Try to get the attribute k of the given object. This may evaluate some
code if the attribute is a property and may fail. In that case, an dict
indicating so is returned.
If get_prop is False, will not return contents of properties (will raise
ValueException).
Determine which attributes of the given object should be enumerated for storage and/or display in UI. Returns a dict of those attributes and their values.
For enumerating contents of objects that do not support utility classes like pydantic, we use this method to guess what should be enumerated when serializing/displaying.
If include_props is True, will produce attributes which are properties;
otherwise those will be excluded.
Threading Utilities¶
Classes¶
Bases: Thread
Thread that wraps target with stack/context tracking.
App components that do not use this thread class might not be properly tracked.
Bases: ThreadPoolExecutor
A ThreadPoolExecutor that keeps track of the stack prior to each thread's invocation.
Apps that do not use this thread pool might not be properly tracked.
Functions¶
Synchronization/Async Utilities¶
NOTE: we cannot name a module "async" as it is a python keyword.
Synchronous vs. Asynchronous¶
Some functions in trulens_eval come with asynchronous versions. Those use "async def" instead of "def" and typically start with the letter "a" in their name with the rest matching their synchronous version.
Due to how python handles such functions and how they are executed, it is
relatively difficult to reshare code between the two versions. Asynchronous
functions are executed by an async loop (see
EventLoop). Python
prevents any threads from having more than one running loop meaning one may not
be able to create one to run some async code if one has already been
created/running in the thread. The method sync here, used to convert an async
computation into a sync computation, needs to create a new thread. The impact of
this, whether overhead, or record info, is uncertain.
What should be Sync/Async?¶
Try to have all internals be async but for users we may expose sync versions via
the sync method. If internals are async and don't need exposure, don't need to
provide a synced version.
Attributes¶
module-attribute
¶
Awaitable or not.
May be checked with isawaitable.
module-attribute
¶
Function or coroutine function.
May be checked with is_really_coroutinefunction.
module-attribute
¶
Function that produces an awaitable / coroutine function.
module-attribute
¶
Thunk or coroutine thunk.
May be checked with is_really_coroutinefunction.
Functions¶
async
¶
desync(func: CallableMaybeAwaitable[A, T], *args, **kwargs) -> T
Run the given function asynchronously with the given args. If it is not
asynchronous, will run in thread. Note: this has to be marked async since in
some cases we cannot tell ahead of time that func is asynchronous so we
may end up running it to produce a coroutine object which we then need to
run asynchronously.
sync(func: CallableMaybeAwaitable[A, T], *args, **kwargs) -> T
Get result of calling function on the given args. If it is awaitable, will block until it is finished. Runs in a new thread in such cases.