Spaces:
Paused
Paused
File size: 4,865 Bytes
4efde5d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 |
"""
JSON helper utilities for handling both legacy (string) and new (dict/list) formats.
These utilities help with the transition from storing JSON as strings to storing
them as proper JSONB objects in the database.
"""
import json
from typing import Any, Union, Dict, List
def ensure_dict(value: Union[str, Dict[str, Any], None], default: Dict[str, Any] = None) -> Dict[str, Any]:
"""
Ensure a value is a dictionary.
Handles:
- None -> returns default or {}
- Dict -> returns as-is
- JSON string -> parses and returns dict
- Other -> returns default or {}
Args:
value: The value to ensure is a dict
default: Default value if conversion fails
Returns:
A dictionary
"""
if default is None:
default = {}
if value is None:
return default
if isinstance(value, dict):
return value
if isinstance(value, str):
try:
parsed = json.loads(value)
if isinstance(parsed, dict):
return parsed
return default
except (json.JSONDecodeError, TypeError):
return default
return default
def ensure_list(value: Union[str, List[Any], None], default: List[Any] = None) -> List[Any]:
"""
Ensure a value is a list.
Handles:
- None -> returns default or []
- List -> returns as-is
- JSON string -> parses and returns list
- Other -> returns default or []
Args:
value: The value to ensure is a list
default: Default value if conversion fails
Returns:
A list
"""
if default is None:
default = []
if value is None:
return default
if isinstance(value, list):
return value
if isinstance(value, str):
try:
parsed = json.loads(value)
if isinstance(parsed, list):
return parsed
return default
except (json.JSONDecodeError, TypeError):
return default
return default
def safe_json_parse(value: Union[str, Dict, List, Any], default: Any = None) -> Any:
"""
Safely parse a value that might be JSON string or already parsed.
This handles the transition period where some data might be stored as
JSON strings (old format) and some as proper objects (new format).
Args:
value: The value to parse
default: Default value if parsing fails
Returns:
Parsed value or default
"""
if value is None:
return default
# If it's already a dict or list, return as-is
if isinstance(value, (dict, list)):
return value
# If it's a string, try to parse it
if isinstance(value, str):
try:
return json.loads(value)
except (json.JSONDecodeError, TypeError):
# If it's not valid JSON, return the string itself
return value
# For any other type, return as-is
return value
def to_json_string(value: Any) -> str:
"""
Convert a value to a JSON string if needed.
This is used for backwards compatibility when yielding data that
expects JSON strings.
Args:
value: The value to convert
Returns:
JSON string representation
"""
if isinstance(value, str):
# If it's already a string, check if it's valid JSON
try:
json.loads(value)
return value # It's already a JSON string
except (json.JSONDecodeError, TypeError):
# It's a plain string, encode it as JSON
return json.dumps(value)
# For all other types, convert to JSON
return json.dumps(value)
def format_for_yield(message_object: Dict[str, Any]) -> Dict[str, Any]:
"""
Format a message object for yielding, ensuring content and metadata are JSON strings.
This maintains backward compatibility with clients expecting JSON strings
while the database now stores proper objects.
Args:
message_object: The message object from the database
Returns:
Message object with content and metadata as JSON strings
"""
if not message_object:
return message_object
# Create a copy to avoid modifying the original
formatted = message_object.copy()
# Ensure content is a JSON string
if 'content' in formatted and not isinstance(formatted['content'], str):
formatted['content'] = json.dumps(formatted['content'])
# Ensure metadata is a JSON string
if 'metadata' in formatted and not isinstance(formatted['metadata'], str):
formatted['metadata'] = json.dumps(formatted['metadata'])
return formatted |