"""Schema transformation utilities for converting data to various schema types."""
import datetime
from collections.abc import Callable, Mapping, Sequence
from decimal import Decimal, InvalidOperation
from enum import Enum
from functools import partial
from pathlib import Path, PurePath
from types import UnionType
from typing import Annotated, Any, Final, TypeGuard, Union, cast, get_args, get_origin, overload
from uuid import UUID
from typing_extensions import TypeVar
from sqlspec.data_dictionary import ForeignKeyMetadata
from sqlspec.exceptions import SQLSpecError
from sqlspec.typing import CATTRS_INSTALLED, NUMPY_INSTALLED, MsgspecValidationError, SchemaT, convert, get_type_adapter
from sqlspec.utils.dispatch import TypeDispatcher
from sqlspec.utils.module_loader import import_optional_attr
from sqlspec.utils.serializers import from_json
from sqlspec.utils.type_guards import (
is_attrs_instance,
is_attrs_schema,
is_dataclass,
is_dict,
is_msgspec_struct,
is_pydantic_model,
is_typed_dict,
)
from sqlspec.utils.uuids import uuid_from_bytes, uuid_from_string
__all__ = (
"_DEFAULT_TYPE_DECODERS",
"ValueT",
"_convert_numpy_recursive",
"_convert_numpy_to_list",
"_default_msgspec_deserializer",
"_is_list_type_target",
"to_schema",
"to_value_type",
"transform_dict_keys",
)
DataT = TypeVar("DataT", default=dict[str, Any])
ValueT = TypeVar("ValueT")
_DATETIME_TYPES: Final[set[type]] = {datetime.datetime, datetime.date, datetime.time}
_DATETIME_TYPE_TUPLE: Final[tuple[type, ...]] = (datetime.datetime, datetime.date, datetime.time)
_MAPPING_TYPE_ARGUMENT_COUNT: Final = 2
_MSGSPEC_FIELD_CACHE: "dict[type, tuple[tuple[tuple[str, str, Any], ...], frozenset[str], frozenset[str]]]" = {}
_NUMPY_RECURSIVE_DISPATCHER: "TypeDispatcher[Callable[[Any], Any]] | None" = None
_NULLABLE_UNION_ARGUMENT_COUNT: Final = 2
@overload
def to_schema(data: "list[DataT]", *, schema_type: "type[SchemaT]") -> "list[SchemaT]": ...
@overload
def to_schema(data: "list[DataT]", *, schema_type: None = None) -> "list[DataT]": ...
@overload
def to_schema(data: "DataT", *, schema_type: "type[SchemaT]") -> "SchemaT": ...
@overload
def to_schema(data: "DataT", *, schema_type: None = None) -> "DataT": ...
[docs]
def to_schema(data: Any, *, schema_type: Any = None) -> Any:
"""Convert data to a specified schema type.
Supports transformation to various schema types including:
- TypedDict
- dataclasses
- msgspec Structs
- Pydantic models
- attrs classes
Args:
data: Input data to convert (dict, list of dicts, or other)
schema_type: Target schema type for conversion. If None, returns data unchanged.
Returns:
Converted data in the specified schema type, or original data if schema_type is None
Raises:
SQLSpecError: If schema_type is not a supported type
"""
if schema_type is None:
return data
conv = _get_schema_converter(schema_type)
if conv is None:
msg = "`schema_type` should be a valid Dataclass, Pydantic model, Msgspec struct, Attrs class, or TypedDict"
raise SQLSpecError(msg)
return conv(data, schema_type)
[docs]
def to_value_type(value: Any, value_type: "type[ValueT]") -> "ValueT":
"""Convert a database value to the specified Python type.
This function handles type conversion for common database return values,
providing runtime type safety for scalar queries. When the value is already
the correct type, it is returned as-is without conversion overhead. Strict
type identity check handles subclass gotchas (bool is subclass of int,
datetime is subclass of date).
Also supports schema types (Pydantic models, dataclasses, msgspec Structs,
attrs classes, and TypedDict). For schema types, JSON strings are automatically
parsed before conversion.
Args:
value: The value to convert.
value_type: The target Python type. Supported types include:
- Primitives: int, float, str, bool
- Temporal: datetime, date, time
- Numeric: Decimal
- Identifiers: UUID, Path
- Collections: dict, list (for JSON/JSONB columns)
- Schema types: Pydantic models, dataclasses, msgspec Structs,
attrs classes, TypedDict (for JSONB columns)
Returns:
The converted value of the specified type.
Raises:
TypeError: If the value cannot be converted to the specified type.
"""
if type(value) is value_type:
return value
if value_type is int:
return cast("ValueT", _convert_to_int(value))
if value_type is str:
return cast("ValueT", str(value))
if value_type is float:
return cast("ValueT", _convert_to_float(value))
if value_type is bool:
return cast("ValueT", _convert_to_bool(value))
if value_type is datetime.datetime:
return cast("ValueT", _convert_to_datetime(value))
if value_type is datetime.date:
return cast("ValueT", _convert_to_date(value))
if value_type is datetime.time:
return cast("ValueT", _convert_to_time(value))
if value_type is Decimal:
return cast("ValueT", _convert_to_decimal(value))
if value_type is UUID:
return cast("ValueT", _convert_to_uuid(value))
if value_type is Path:
return cast("ValueT", _convert_to_path(value))
if value_type is dict:
return cast("ValueT", _convert_to_dict(value))
if value_type is list:
return cast("ValueT", _convert_to_list(value))
schema_converter = _get_schema_converter(value_type)
if schema_converter is not None:
parsed = _ensure_json_parsed(value)
return cast("ValueT", schema_converter(parsed, value_type))
try:
return value_type(value) # type: ignore[call-arg]
except (TypeError, ValueError) as e:
msg = f"Cannot convert {type(value).__name__} to {value_type.__name__}"
raise TypeError(msg) from e
def _safe_convert_key(key: Any, converter: Callable[[str], str]) -> Any:
"""Safely convert a key using the converter function.
Args:
key: Key to convert (may not be a string).
converter: Function to convert string keys.
Returns:
Converted key if conversion succeeds, original key otherwise.
"""
if not isinstance(key, str):
return key
try:
return converter(key)
except (TypeError, ValueError, AttributeError):
return key
def _transform_dict(data: dict, converter: Callable[[str], str]) -> dict:
"""Transform a dictionary's keys recursively.
Args:
data: Dictionary to transform.
converter: Function to convert string keys.
Returns:
Dictionary with transformed keys and recursively transformed values.
"""
transformed = {}
for key, value in data.items():
converted_key = _safe_convert_key(key, converter)
transformed_value = transform_dict_keys(value, converter)
transformed[converted_key] = transformed_value
return transformed
def _transform_list(data: list, converter: Callable[[str], str]) -> list:
"""Transform a list's elements recursively.
Args:
data: List to transform.
converter: Function to convert string keys in nested structures.
Returns:
List with recursively transformed elements.
"""
return [transform_dict_keys(item, converter) for item in data]
def _is_list_type_target(target_type: Any) -> "TypeGuard[list[object]]":
"""Check if target type is a list type."""
try:
origin = target_type.__origin__
except (AttributeError, TypeError):
return False
return origin is list
def _convert_numpy_to_list(target_type: Any, value: Any) -> Any:
"""Convert numpy array to list if target is a list type."""
if not NUMPY_INSTALLED:
return value
import numpy as np
if isinstance(value, np.ndarray) and _is_list_type_target(target_type):
return value.tolist()
return value
def _convert_foreign_key_metadata(data: Any, schema_type: Any) -> Any:
"""Convert data to ForeignKeyMetadata schema type.
Handles both single items and lists for consistency with other converters.
"""
if isinstance(data, list):
return [_coerce_foreign_key_item(item, schema_type) for item in data]
return _coerce_foreign_key_item(data, schema_type)
def _coerce_foreign_key_item(data: Any, schema_type: Any) -> Any:
"""Convert a single item to ForeignKeyMetadata."""
if not is_dict(data):
return data
payload = {
"table_name": data.get("table_name") or data.get("table"),
"column_name": data.get("column_name") or data.get("column"),
"referenced_table": data.get("referenced_table") or data.get("referenced_table_name"),
"referenced_column": data.get("referenced_column") or data.get("referenced_column_name"),
"constraint_name": data.get("constraint_name"),
"schema": data.get("schema") or data.get("table_schema"),
"referenced_schema": data.get("referenced_schema") or data.get("referenced_table_schema"),
}
return schema_type(**payload)
def _convert_typed_dict(data: Any, schema_type: Any) -> Any:
"""Convert data to TypedDict."""
return data
def _convert_dataclass(data: Any, schema_type: Any) -> Any:
"""Convert data to dataclass."""
if isinstance(data, list):
return [schema_type(**item) if is_dict(item) else item for item in data]
return schema_type(**data) if is_dict(data) else data
class _IsTypePredicate:
"""Callable predicate to check if a type matches a target type."""
__slots__ = ("_type",)
def __init__(self, target_type: type) -> None:
self._type = target_type
def __call__(self, x: Any) -> bool:
return x is self._type
class _UUIDDecoder:
"""Decoder for UUID types."""
__slots__ = ()
def __call__(self, t: type, v: Any) -> Any:
return t(v.hex)
class _ISOFormatDecoder:
"""Decoder for types with isoformat() method (datetime, date, time)."""
__slots__ = ()
def __call__(self, t: type, v: Any) -> Any:
return t(v.isoformat())
class _EnumDecoder:
"""Decoder for Enum types."""
__slots__ = ()
def __call__(self, t: type, v: Any) -> Any:
return t(v.value)
_DEFAULT_TYPE_DECODERS: Final[list[tuple[Callable[[Any], bool], Callable[[Any, Any], Any]]]] = [
(_IsTypePredicate(UUID), _UUIDDecoder()),
(_IsTypePredicate(datetime.datetime), _ISOFormatDecoder()),
(_IsTypePredicate(datetime.date), _ISOFormatDecoder()),
(_IsTypePredicate(datetime.time), _ISOFormatDecoder()),
(_IsTypePredicate(Enum), _EnumDecoder()),
(_is_list_type_target, _convert_numpy_to_list),
]
def _default_msgspec_deserializer(
target_type: Any, value: Any, type_decoders: "Sequence[tuple[Any, Any]] | None" = None
) -> Any:
"""Convert msgspec types with type decoder support.
Args:
target_type: Type to convert to
value: Value to convert
type_decoders: Optional sequence of (predicate, decoder) pairs
Returns:
Converted value or original value if conversion not applicable
"""
if NUMPY_INSTALLED:
import numpy as np
if isinstance(value, np.ndarray) and _is_list_type_target(target_type):
return value.tolist()
if type_decoders:
for predicate, decoder in type_decoders:
if predicate(target_type):
return decoder(target_type, value)
if target_type is UUID and isinstance(value, UUID):
return value.hex
if target_type in _DATETIME_TYPES and isinstance(value, _DATETIME_TYPE_TUPLE):
datetime_value = cast("datetime.datetime | datetime.date | datetime.time", value)
return datetime_value.isoformat()
if isinstance(target_type, type) and issubclass(target_type, Enum) and isinstance(value, Enum):
return value.value
try:
if isinstance(target_type, type) and isinstance(value, target_type):
return value
except TypeError:
pass
if isinstance(target_type, type):
try:
if issubclass(target_type, (Path, PurePath)) or issubclass(target_type, UUID):
return target_type(str(value))
except (TypeError, ValueError):
pass
return value
_DEFAULT_MSGSPEC_DESERIALIZER: Final[Callable[[Any, Any], Any]] = partial(
_default_msgspec_deserializer, type_decoders=_DEFAULT_TYPE_DECODERS
)
def _convert_numpy_recursive(obj: Any) -> Any:
"""Recursively convert numpy arrays to lists.
This is a module-level function to avoid nested function definitions
which are problematic for mypyc compilation.
Args:
obj: Object to convert (may contain numpy arrays nested in dicts/lists)
Returns:
Object with all numpy arrays converted to lists
"""
if not NUMPY_INSTALLED:
return obj
handler = _get_numpy_recursive_dispatcher().get(obj)
if handler is not None:
return handler(obj)
return obj
def _convert_numpy_array(obj: Any) -> Any:
return obj.tolist()
def _convert_numpy_mapping(obj: Any) -> Any:
return {key: _convert_numpy_recursive(value) for key, value in obj.items()}
def _convert_numpy_sequence(obj: Any) -> Any:
converted = [_convert_numpy_recursive(item) for item in obj]
return type(obj)(converted)
def _get_numpy_recursive_dispatcher() -> "TypeDispatcher[Callable[[Any], Any]]":
global _NUMPY_RECURSIVE_DISPATCHER
if _NUMPY_RECURSIVE_DISPATCHER is None:
import numpy as np
dispatcher = TypeDispatcher["Callable[[Any], Any]"]()
dispatcher.register(np.ndarray, _convert_numpy_array)
dispatcher.register(dict, _convert_numpy_mapping)
dispatcher.register(list, _convert_numpy_sequence)
dispatcher.register(tuple, _convert_numpy_sequence)
_NUMPY_RECURSIVE_DISPATCHER = dispatcher
return _NUMPY_RECURSIVE_DISPATCHER
def _convert_msgspec(data: Any, schema_type: Any) -> Any:
"""Convert data to msgspec Struct."""
transformed_data = (
[_normalize_msgspec_input(item, schema_type) for item in data]
if isinstance(data, Sequence) and not isinstance(data, (str, bytes, bytearray))
else _normalize_msgspec_input(data, schema_type)
)
target_type = list[schema_type] if isinstance(transformed_data, Sequence) else schema_type
try:
return convert(
obj=transformed_data, type=target_type, from_attributes=True, dec_hook=_DEFAULT_MSGSPEC_DESERIALIZER
)
except MsgspecValidationError:
if not NUMPY_INSTALLED:
raise
transformed_data = _convert_numpy_recursive(transformed_data)
return convert(
obj=transformed_data, type=target_type, from_attributes=True, dec_hook=_DEFAULT_MSGSPEC_DESERIALIZER
)
def _normalize_msgspec_input(data: Any, target_type: Any) -> Any:
"""Normalize Struct field aliases according to the declared target type."""
target_type = _unwrap_msgspec_target(target_type)
if target_type is None:
return data
if is_msgspec_struct(target_type):
return _normalize_msgspec_struct(data, cast("type", target_type))
origin = get_origin(target_type)
args = get_args(target_type)
if origin is None or not args:
return data
if _is_mapping_origin(origin):
if not isinstance(data, Mapping) or len(args) < _MAPPING_TYPE_ARGUMENT_COUNT:
return data
value_type = args[1]
return {key: _normalize_msgspec_input(value, value_type) for key, value in data.items()}
if _is_sequence_origin(origin):
if not isinstance(data, Sequence) or isinstance(data, (str, bytes, bytearray)):
return data
if origin is tuple and len(args) > 1 and args[1] is not Ellipsis:
return [
_normalize_msgspec_input(value, args[index]) if index < len(args) else value
for index, value in enumerate(data)
]
item_type = next(iter(args), Any)
return [_normalize_msgspec_input(value, item_type) for value in data]
return data
def _normalize_msgspec_struct(data: Any, schema_type: type) -> Any:
if not isinstance(data, Mapping):
return data
fields, python_names, encoded_names = _msgspec_field_plan(schema_type)
normalized = {key: value for key, value in data.items() if key not in python_names and key not in encoded_names}
for name, encode_name, field_type in fields:
if encode_name in data:
normalized[encode_name] = _normalize_msgspec_input(data[encode_name], field_type)
elif name in data:
normalized[encode_name] = _normalize_msgspec_input(data[name], field_type)
return normalized
def _msgspec_field_plan(schema_type: type) -> "tuple[tuple[tuple[str, str, Any], ...], frozenset[str], frozenset[str]]":
try:
return _MSGSPEC_FIELD_CACHE[schema_type]
except KeyError:
from msgspec import structs
fields = tuple(
(field.name, field.encode_name, field.type) for field in structs.fields(cast("Any", schema_type))
)
plan = fields, frozenset(field[0] for field in fields), frozenset(field[1] for field in fields)
_MSGSPEC_FIELD_CACHE[schema_type] = plan
return plan
def _unwrap_msgspec_target(target_type: Any) -> Any:
while get_origin(target_type) is Annotated:
target_type = get_args(target_type)[0]
origin = get_origin(target_type)
if origin is Union or origin is UnionType:
args = get_args(target_type)
non_null = tuple(arg for arg in args if arg is not type(None))
if len(args) == _NULLABLE_UNION_ARGUMENT_COUNT and len(non_null) == 1:
return _unwrap_msgspec_target(non_null[0])
return None
return target_type
def _is_mapping_origin(origin: Any) -> bool:
try:
return issubclass(origin, Mapping)
except TypeError:
return False
def _is_sequence_origin(origin: Any) -> bool:
try:
return issubclass(origin, Sequence)
except TypeError:
return False
def _convert_pydantic(data: Any, schema_type: Any) -> Any:
"""Convert data to Pydantic model."""
if isinstance(data, Sequence):
return get_type_adapter(list[schema_type]).validate_python(data, from_attributes=True)
return get_type_adapter(schema_type).validate_python(data, from_attributes=True)
def _convert_attrs(data: Any, schema_type: Any) -> Any:
"""Convert data to attrs class."""
if CATTRS_INSTALLED:
cattrs_structure = import_optional_attr("cattrs", "structure")
cattrs_unstructure = import_optional_attr("cattrs", "unstructure")
if isinstance(data, Sequence):
return cattrs_structure(data, list[schema_type])
structured = cattrs_unstructure(data) if is_attrs_instance(data) else data
return cattrs_structure(structured, schema_type)
if isinstance(data, list):
attrs_asdict = import_optional_attr("attrs", "asdict")
return [schema_type(**item) if is_dict(item) else schema_type(**attrs_asdict(item)) for item in data]
return schema_type(**data) if is_dict(data) else data
_SCHEMA_CONVERTER_CACHE: "dict[type, Callable[[Any, Any], Any] | None]" = {}
def _get_schema_converter(schema_type: type) -> "Callable[[Any, Any], Any] | None":
"""Get the converter function for a schema type with caching.
This is optimized for mypyc: uses a plain dict cache instead of lru_cache,
and caches the callable directly instead of a string key.
Args:
schema_type: The schema type to get a converter for.
Returns:
Converter function if schema_type is a supported schema, None otherwise.
"""
try:
return _SCHEMA_CONVERTER_CACHE[schema_type]
except KeyError:
if is_typed_dict(schema_type):
conv: Callable[[Any, Any], Any] | None = _convert_typed_dict
elif is_dataclass(schema_type):
conv = _convert_dataclass
elif is_msgspec_struct(schema_type):
conv = _convert_msgspec
elif is_pydantic_model(schema_type):
conv = _convert_pydantic
elif is_attrs_schema(schema_type):
conv = _convert_attrs
elif isinstance(schema_type, type) and issubclass(schema_type, ForeignKeyMetadata):
conv = _convert_foreign_key_metadata
else:
conv = None
_SCHEMA_CONVERTER_CACHE[schema_type] = conv # pyright: ignore[reportArgumentType]
return conv
def _ensure_json_parsed(value: Any) -> Any:
"""Parse JSON string if needed, otherwise return as-is.
This helper is used when converting database values (potentially JSON strings
from JSONB columns) to schema types like Pydantic models or dataclasses.
Args:
value: The value to potentially parse. If it's a string, attempts JSON parsing.
Returns:
Parsed JSON object if value was a valid JSON string, otherwise the original value.
"""
if isinstance(value, str):
try:
return from_json(value)
except Exception:
return value
return value
def _try_parse_json(value: str) -> Any:
"""Attempt to parse a JSON string, returning None on failure.
Args:
value: JSON string to parse.
Returns:
Parsed JSON value, or None if parsing fails.
"""
try:
return from_json(value)
except Exception:
return None
_BOOL_TRUE_VALUES: Final[frozenset[str]] = frozenset({"true", "1", "yes", "y", "t", "on"})
_STRICT_IDENTITY_TYPES: Final[tuple[type, ...]] = (int, bool, datetime.date, datetime.time)
def _convert_to_int(value: Any) -> int:
"""Convert a value to int.
Args:
value: Value to convert.
Returns:
Converted integer value.
Raises:
TypeError: If value cannot be converted to int.
"""
if isinstance(value, bool):
return int(value)
if isinstance(value, (int, float, Decimal)):
return int(value)
if isinstance(value, str):
try:
return int(value)
except ValueError:
try:
return int(float(value))
except ValueError:
pass
msg = f"Cannot convert {type(value).__name__} to int"
raise TypeError(msg)
def _convert_to_float(value: Any) -> float:
"""Convert a value to float.
Args:
value: Value to convert.
Returns:
Converted float value.
Raises:
TypeError: If value cannot be converted to float.
"""
if isinstance(value, bool):
return float(value)
if isinstance(value, (int, float, Decimal)):
return float(value)
if isinstance(value, str):
try:
return float(value)
except ValueError:
pass
msg = f"Cannot convert {type(value).__name__} to float"
raise TypeError(msg)
def _convert_to_bool(value: Any) -> bool:
"""Convert a value to bool.
Args:
value: Value to convert.
Returns:
Converted boolean value.
Raises:
TypeError: If value cannot be converted to bool.
"""
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
if isinstance(value, str):
return value.lower() in _BOOL_TRUE_VALUES
msg = f"Cannot convert {type(value).__name__} to bool"
raise TypeError(msg)
def _convert_to_datetime(value: Any) -> datetime.datetime:
"""Convert a value to datetime.
Args:
value: Value to convert.
Returns:
Converted datetime value.
Raises:
TypeError: If value cannot be converted to datetime.
"""
if isinstance(value, datetime.datetime):
return value
if isinstance(value, str):
try:
return datetime.datetime.fromisoformat(value)
except ValueError:
pass
if isinstance(value, datetime.date) and not isinstance(value, datetime.datetime):
return datetime.datetime.combine(value, datetime.time.min)
msg = f"Cannot convert {type(value).__name__} to datetime"
raise TypeError(msg)
def _convert_to_date(value: Any) -> datetime.date:
"""Convert a value to date.
Args:
value: Value to convert.
Returns:
Converted date value.
Raises:
TypeError: If value cannot be converted to date.
"""
if isinstance(value, datetime.datetime):
return value.date()
if isinstance(value, datetime.date):
return value
if isinstance(value, str):
try:
return datetime.date.fromisoformat(value)
except ValueError:
try:
return datetime.datetime.fromisoformat(value).date()
except ValueError:
pass
msg = f"Cannot convert {type(value).__name__} to date"
raise TypeError(msg)
def _convert_to_time(value: Any) -> datetime.time:
"""Convert a value to time.
Args:
value: Value to convert.
Returns:
Converted time value.
Raises:
TypeError: If value cannot be converted to time.
"""
if isinstance(value, datetime.datetime):
return value.time()
if isinstance(value, datetime.time):
return value
if isinstance(value, str):
try:
return datetime.time.fromisoformat(value)
except ValueError:
pass
msg = f"Cannot convert {type(value).__name__} to time"
raise TypeError(msg)
def _convert_to_decimal(value: Any) -> Decimal:
"""Convert a value to Decimal.
Args:
value: Value to convert.
Returns:
Converted Decimal value.
Raises:
TypeError: If value cannot be converted to Decimal.
"""
if isinstance(value, Decimal):
return value
if isinstance(value, (int, float, str)):
try:
return Decimal(str(value))
except InvalidOperation:
pass
msg = f"Cannot convert {type(value).__name__} to Decimal"
raise TypeError(msg)
def _convert_to_uuid(value: Any) -> UUID:
"""Convert a value to UUID.
Args:
value: Value to convert.
Returns:
Converted UUID value.
Raises:
TypeError: If value cannot be converted to UUID.
"""
if isinstance(value, UUID):
return value
if isinstance(value, str):
try:
return uuid_from_string(value)
except ValueError:
pass
if isinstance(value, bytes):
try:
return uuid_from_bytes(value)
except ValueError:
pass
msg = f"Cannot convert {type(value).__name__} to UUID"
raise TypeError(msg)
def _convert_to_path(value: Any) -> Path:
"""Convert a value to Path.
Args:
value: Value to convert.
Returns:
Converted Path value.
Raises:
TypeError: If value cannot be converted to Path.
"""
if isinstance(value, Path):
return value
if isinstance(value, (str, PurePath)):
return Path(value)
msg = f"Cannot convert {type(value).__name__} to Path"
raise TypeError(msg)
def _convert_to_dict(value: Any) -> dict[str, Any]:
"""Convert a value to dict.
This is useful for JSON/JSONB database columns where the driver may return
either a dict (already parsed) or a string (needs parsing).
Args:
value: Value to convert.
Returns:
Converted dict value.
Raises:
TypeError: If value cannot be converted to dict.
"""
if isinstance(value, dict):
return value
if isinstance(value, str):
parsed = _try_parse_json(value)
if parsed is not None:
if isinstance(parsed, dict):
return parsed
msg = f"JSON string did not parse to dict, got {type(parsed).__name__}"
raise TypeError(msg)
msg = f"Cannot convert {type(value).__name__} to dict"
raise TypeError(msg)
def _convert_to_list(value: Any) -> list[Any]:
"""Convert a value to list.
This is useful for JSON array database columns where the driver may return
either a list (already parsed) or a string (needs parsing).
Args:
value: Value to convert.
Returns:
Converted list value.
Raises:
TypeError: If value cannot be converted to list.
"""
if isinstance(value, list):
return value
if isinstance(value, str):
parsed = _try_parse_json(value)
if parsed is not None:
if isinstance(parsed, list):
return parsed
msg = f"JSON string did not parse to list, got {type(parsed).__name__}"
raise TypeError(msg)
if isinstance(value, (tuple, set, frozenset)):
return list(value)
msg = f"Cannot convert {type(value).__name__} to list"
raise TypeError(msg)
