Shared Functionality

Shared functionality for general use.

This module contains shared functionality for general use.

WeakAsyncLockRegistry(lock_factory=None)

Bases: Generic[KeyT]

Store asyncio locks keyed by objects using weak references.

Locks are created lazily to avoid requiring an active event loop at construction time.

Initialize the registry with an optional custom lock factory.

Parameters:

Name	Type	Description	Default
lock_factory	Callable[[], Lock] | None	The factory to create locks.	None

get(key)

Return the shared lock for the given key, creating it if absent.

Parameters:

Name	Type	Description	Default
key	KeyT	The key to get the lock for.	required

Returns:

Type	Description
Lock	asyncio.Lock: The shared lock for the given key.

convert_traces_to_data_rows(traces, export_mapping=None)

Convert Langfuse traces to data rows based on export mapping.

Parameters:

Name	Type	Description	Default
traces	list[dict[str, Any]]	List of trace dictionaries from Langfuse.	required
export_mapping	dict[str, str] | None	Mapping from trace keys to CSV column names. If None, uses EXPORT_MAPPING from constants. Defaults to None.	None

Returns:

Type	Description
tuple[list[dict[str, Any]], list[str]]	tuple[list[dict[str, Any]], list[str]]: Tuple of (list of row dictionaries, ordered column names).

extract_dataset_name_from_path(path, file_extension)

Extract dataset name from file path by removing directory and extension.

Parameters:

Name	Type	Description	Default
path	str	The file path (e.g., "/path/to/data.csv", "data.jsonl")	required
file_extension	str	The file extension to remove (e.g., ".csv", ".jsonl")	required

Returns:

Name	Type	Description
str	str	The extracted dataset name (e.g., "data")

Examples:

>>> extract_dataset_name_from_path("/path/to/my_data.csv", ".csv")
"my_data"
>>> extract_dataset_name_from_path("evaluation_set.jsonl", ".jsonl")
"evaluation_set"

extract_metrics_with_fallback(data, gt_prefix='gt_', predicted_prefix='predicted_')

Extract metrics starting with gt_prefix, falling back to predicted_prefix if gt is None.

This function finds all keys starting with gt_prefix, and for each metric: - If the gt_ value is not None, use it - If the gt_ value is None, check for the corresponding predicted_ value - If predicted_ value exists and is not None, use it instead - The returned dictionary uses metric names without the prefix

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The dictionary containing metrics.	required
gt_prefix	str	Prefix for ground truth metrics. Defaults to "gt_".	'gt_'
predicted_prefix	str	Prefix for predicted metrics. Defaults to "predicted_".	'predicted_'

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Dictionary with metric names (without prefix) as keys and their values.

Examples:

Basic usage with fallback: >>> data = { ... "gt_completeness": None, ... "predicted_completeness": 3.0, ... "gt_redundancy": 2.0, ... "predicted_redundancy": 1.0 ... } >>> extract_metrics_with_fallback(data)

All gt values present: >>> data = { ... "gt_completeness": 3.0, ... "gt_redundancy": 2.0 ... } >>> extract_metrics_with_fallback(data)

All gt values None, using predicted: >>> data = { ... "gt_completeness": None, ... "predicted_completeness": 3.0, ... "gt_generation": None, ... "predicted_generation": 0 ... } >>> extract_metrics_with_fallback(data)

Mixed case: >>> data = { ... "gt_completeness": None, ... "predicted_completeness": 3.0, ... "gt_generation": None, ... "predicted_generation": 0, ... "gt_redundancy": 2.0, ... "predicted_redundancy": 1.0 ... } >>> extract_metrics_with_fallback(data)

generate_dataset_name_with_timestamp(base_name='dataset')

Generate a dataset name with timestamp postfix.

The timestamp is in compact ISO 8601 format: YYYYMMDDTHHMMSSZ

Parameters:

Name	Type	Description	Default
base_name	str	The base name for the dataset. Defaults to "dataset".	'dataset'

Returns:

Name	Type	Description
str	str	The dataset name with timestamp postfix.

Example

generate_dataset_name_with_timestamp("my_dataset") 'my_dataset_20260420T143000Z'

generate_run_id_if_not_provided(project_name, dataset_name, run_id=None)

Generate a run ID.

Parameters:

Name	Type	Description	Default
project_name	str	The name of the project.	required
dataset_name	str	The name of the dataset.	required
run_id	str | None	Provided run id or None.	None

Returns:

Name	Type	Description
str	str	The run ID.

get_nested_dict_value(data, key, return_all=False)

Recursively get the value of a key from a nested dictionary.

This function searches through nested dictionaries and lists containing dictionaries to find the specified key and return its value(s).

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The dictionary to search in.	required
key	str	The key to search for.	required
return_all	bool	If True, returns all matching values as a list. If False, returns the first match found. Defaults to False.	False

Returns:

Type	Description
Any | list[Any] | None	Any | list[Any] | None: The value(s) found for the key. Returns None if key is not found (when return_all=False), or an empty list if not found (when return_all=True).

Examples:

Basic nested dictionary: >>> data = {"level1": {"level2": {"target": "found"}}} >>> get_nested_dict_value(data, "target") "found"

Key at multiple levels (returns first match): >>> data = {"target": "first", "nested": {"target": "second"}} >>> get_nested_dict_value(data, "target") "first"

Get all matches: >>> data = {"target": "first", "nested": {"target": "second"}} >>> get_nested_dict_value(data, "target", return_all=True) ["first", "second"]

List of dictionaries: >>> data = {"items": [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}]} >>> get_nested_dict_value(data, "name") "A"

Complex nested structure: >>> data = { ... "config": {"settings": {"timeout": 30}}, ... "users": [{"name": "Alice", "role": "admin"}, {"name": "Bob"}], ... "metadata": {"tags": ["important"], "timeout": 60} ... } >>> get_nested_dict_value(data, "timeout") 30 >>> get_nested_dict_value(data, "timeout", return_all=True) [30, 60]

Key not found: >>> get_nested_dict_value({"a": 1}, "missing") None >>> get_nested_dict_value({"a": 1}, "missing", return_all=True) []

map_results(results)

Map the results to the expected format.

Parameters:

Name	Type	Description	Default
results	dict[str, Any]	The results to map.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: The mapped results.

parse_list_strings_in_dict(data)

Parse string representations of lists in dictionary values.

Recursively processes all values in a dictionary, converting string representations of lists (e.g., '["abc", "def"]') into actual Python lists. Also handles CSV-escaped JSON strings (e.g., '[""abc"", ""def""]').

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary to process	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Dictionary with parsed list strings converted to actual lists

Examples:

Basic string to list conversion: >>> parse_list_strings_in_dict({"key1": "abc", "key2": '["item1", "item2"]'})

Nested dictionary processing: >>> parse_list_strings_in_dict({"nested": {"list": '["a", "b"]'}}) {"nested": {"list": ["a", "b"]}}

CSV-escaped JSON handling: >>> parse_list_strings_in_dict({"csv_escaped": '[""item1"", ""item2""]'})

Mixed data types: >>> parse_list_strings_in_dict({"str": "hello", "list": '["a", "b"]', "num": 42, "bool": True})

List of dictionaries with nested parsing: >>> parse_list_strings_in_dict({"items": [{"name": '["item1", "item2"]'}, {"id": 123}]}) {"items": [{"name": ["item1", "item2"]}, {"id": 123}]}

Complex nested structure: >>> data = { ... "config": {"features": '["auth", "logging"]'}, ... "users": [{"roles": '["admin", "user"]'}, {"status": "active"}], ... "settings": {"enabled": True, "timeout": 30} ... } >>> parse_list_strings_in_dict(data) { "config": {"features": ["auth", "logging"]}, "users": [{"roles": ["admin", "user"]}, {"status": "active"}], "settings": {"enabled": True, "timeout": 30} }

Single-quoted strings (AST parsing): >>> parse_list_strings_in_dict({"single_quotes": "['a', 'b', 'c']"})

Non-list strings (unchanged): >>> parse_list_strings_in_dict({"not_list": "just a string", "empty": ""})

write_traces_to_csv(rows, ordered_columns, output_path)

Write Langfuse traces to a CSV file.

Parameters:

Name	Type	Description	Default
rows	list[dict[str, Any]]	List of row dictionaries.	required
ordered_columns	list[str]	List of ordered column names.	required
output_path	str	Path to the output CSV file.	required

Returns:

Type	Description
None	None