Skip to content

trulens.core.utils.containers

trulens.core.utils.containers

Container class utilities.

Classes

BlockingSet

Bases: set, Generic[T]

A set with max size that has blocking peek/get/add .

Functions
empty
empty() -> bool

Check if the set is empty.

shutdown
shutdown()

Shutdown the set.

peek
peek() -> T

Get an item from the set.

Blocks until an item is available.

remove
remove(item: T)

Remove an item from the set.

pop
pop(blocking: bool = True) -> Optional[T]

Get and remove an item from the set.

Blocks until an item is available, unless blocking is set to False.

PARAMETER DESCRIPTION
blocking

Whether to block until an item is ready. If not blocking and empty, will return None.

TYPE: bool DEFAULT: True

add
add(item: T)

Add an item to the set.

Blocks if set is full.

Functions

datetime_of_ns_timestamp

datetime_of_ns_timestamp(timestamp: int) -> datetime

Convert a nanosecond timestamp to a datetime.

ns_timestamp_of_datetime

ns_timestamp_of_datetime(dt: datetime) -> int

Convert a datetime to a nanosecond timestamp.

first

first(seq: Sequence[T]) -> T

Get the first item in a sequence.

second

second(seq: Sequence[T]) -> T

Get the second item in a sequence.

third

third(seq: Sequence[T]) -> T

Get the third item in a sequence.

is_empty

is_empty(obj)

Check if an object is empty.

If object is not a sequence, returns False.

dict_set_with

dict_set_with(
    dict1: Dict[A, B], dict2: Dict[A, B]
) -> Dict[A, B]

Add the key/values from dict2 to dict1.

Mutates and returns dict1.

dict_set_with_multikey

dict_set_with_multikey(
    dict1: Dict[A, B],
    dict2: Dict[Union[A, Tuple[A, ...]], B],
) -> Dict[A, B]

Like dict_set_with except the second dict can have tuples as keys in which case all of the listed keys are set to the given value.

dict_merge_with

dict_merge_with(
    dict1: Dict, dict2: Dict, merge: Callable
) -> Dict

Merge values from the second dictionary into the first.

If both dicts contain the same key, the given merge function is used to merge the values.