backstory/src/backend/helpers/check_serializable.py

60 lines
2.4 KiB
Python

from pydantic import BaseModel
import json
from typing import Any, List, Set
def check_serializable(obj: Any, path: str = "", errors: List[str] = [], visited: Set[int] = set()) -> List[str]:
"""
Recursively check all fields in an object for non-JSON-serializable types, avoiding infinite recursion.
Skips fields in Pydantic models marked with Field(..., exclude=True).
Args:
obj: The object to inspect (Pydantic model, dict, list, or other).
path: The current field path (e.g., 'field1.nested_field').
errors: List to collect error messages.
visited: Set of object IDs to track visited objects and prevent infinite recursion.
Returns:
List of error messages for non-serializable fields.
"""
# Check for circular reference by object ID
obj_id = id(obj)
if obj_id in visited:
errors.append(f"Field '{path}' contains a circular reference, skipping further inspection")
return errors
# Add current object to visited set
visited.add(obj_id)
try:
# Handle Pydantic models
if isinstance(obj, BaseModel):
for field_name, field_info in obj.model_fields.items():
# Skip fields marked with exclude=True
if field_info.exclude:
continue
value = getattr(obj, field_name)
new_path = f"{path}.{field_name}" if path else field_name
check_serializable(value, new_path, errors, visited)
# Handle dictionaries
elif isinstance(obj, dict):
for key, value in obj.items():
new_path = f"{path}[{key}]" if path else str(key)
check_serializable(value, new_path, errors, visited)
# Handle lists, tuples, or other iterables
elif isinstance(obj, (list, tuple)):
for i, value in enumerate(obj):
new_path = f"{path}[{i}]" if path else str(i)
check_serializable(value, new_path, errors, visited)
# Handle other types (check for JSON serializability)
else:
try:
json.dumps(obj)
except (TypeError, OverflowError, ValueError) as e:
errors.append(f"Field '{path}' contains non-serializable type: {type(obj)} ({str(e)})")
finally:
# Remove the current object from visited to allow processing in other branches
visited.discard(obj_id)
return errors