Coverage for haystack/tracing/utils.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import json 

6from typing import Any 

7 

8from haystack import logging 

9 

10logger = logging.getLogger(__name__) 

11 

12PRIMITIVE_TYPES = (bool, str, int, float) 

13 

14 

15def coerce_tag_value(value: Any) -> bool | str | int | float: 

16 """ 

17 Coerces span tag values to compatible types for the tracing backend. 

18 

19 Most tracing libraries don't support sending complex types to the backend. Hence, we need to convert them to 

20 compatible types. 

21 

22 :param value: an arbitrary value which should be coerced to a compatible type 

23 :return: the value coerced to a compatible type 

24 """ 

25 if isinstance(value, PRIMITIVE_TYPES): 

26 return value 

27 

28 if value is None: 

29 return "" 

30 

31 try: 

32 # do that with-in try-except because who knows what kind of objects are being passed 

33 serializable = _serializable_value(value=value, use_placeholders=True) 

34 return json.dumps(serializable) 

35 except Exception as error: 

36 logger.debug("Failed to coerce tag value to string: {error}", error=error) 

37 

38 # Our last resort is to convert the value to a string 

39 return str(value) 

40 

41 

42def _serializable_value(value: Any, use_placeholders: bool = True) -> Any: 

43 """ 

44 Serializes a value into a format suitable for tracing. 

45 

46 One-way only: this is never deserialized, so it's allowed to lose data (e.g. replace large 

47 objects with a placeholder). Don't swap it for `base_serialization`'s schema serializer, which 

48 is built to round-trip exactly. 

49 

50 :param value: 

51 The value to serialize. 

52 :param use_placeholders: 

53 Whether to use string placeholders for large objects like ByteStream and ImageContent. 

54 :returns: 

55 The serialized value. 

56 """ 

57 if isinstance(value, list): 

58 return [_serializable_value(value=v, use_placeholders=use_placeholders) for v in value] 

59 

60 if isinstance(value, dict): 

61 return {k: _serializable_value(value=v, use_placeholders=use_placeholders) for k, v in value.items()} 

62 

63 if use_placeholders and getattr(value, "_to_trace_dict", None): 

64 return _serializable_value(value._to_trace_dict()) 

65 

66 if getattr(value, "to_dict", None): 

67 return _serializable_value(value.to_dict()) 

68 

69 return value