forked from phoenix-oss/llama-stack-mirror
ec4fc800cc
This is yet another of those large PRs (hopefully we will have less and less of them as things mature fast). This one introduces substantial improvements and some simplifications to the stack. Most important bits: * Agents reference implementation now has support for session / turn persistence. The default implementation uses sqlite but there's also support for using Redis. * We have re-architected the structure of the Stack APIs to allow for more flexible routing. The motivating use cases are: - routing model A to ollama and model B to a remote provider like Together - routing shield A to local impl while shield B to a remote provider like Bedrock - routing a vector memory bank to Weaviate while routing a keyvalue memory bank to Redis * Support for provider specific parameters to be passed from the clients. A client can pass data using `x_llamastack_provider_data` parameter which can be type-checked and provided to the Adapter implementations.
959 lines
33 KiB
Python
959 lines
33 KiB
Python
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
|
# All rights reserved.
|
|
#
|
|
# This source code is licensed under the terms described in the LICENSE file in
|
|
# the root directory of this source tree.
|
|
|
|
"""
|
|
Type-safe data interchange for Python data classes.
|
|
|
|
:see: https://github.com/hunyadi/strong_typing
|
|
"""
|
|
|
|
import abc
|
|
import base64
|
|
import dataclasses
|
|
import datetime
|
|
import enum
|
|
import inspect
|
|
import ipaddress
|
|
import sys
|
|
import typing
|
|
import uuid
|
|
from types import ModuleType
|
|
from typing import (
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
Generic,
|
|
List,
|
|
Literal,
|
|
NamedTuple,
|
|
Optional,
|
|
Set,
|
|
Tuple,
|
|
Type,
|
|
TypeVar,
|
|
Union,
|
|
)
|
|
|
|
from .core import JsonType
|
|
from .exception import JsonKeyError, JsonTypeError, JsonValueError
|
|
from .inspection import (
|
|
create_object,
|
|
enum_value_types,
|
|
evaluate_type,
|
|
get_class_properties,
|
|
get_class_property,
|
|
get_resolved_hints,
|
|
is_dataclass_instance,
|
|
is_dataclass_type,
|
|
is_named_tuple_type,
|
|
is_type_annotated,
|
|
is_type_literal,
|
|
is_type_optional,
|
|
TypeLike,
|
|
unwrap_annotated_type,
|
|
unwrap_literal_values,
|
|
unwrap_optional_type,
|
|
)
|
|
from .mapping import python_field_to_json_property
|
|
from .name import python_type_to_str
|
|
|
|
E = TypeVar("E", bound=enum.Enum)
|
|
T = TypeVar("T")
|
|
R = TypeVar("R")
|
|
K = TypeVar("K")
|
|
V = TypeVar("V")
|
|
|
|
|
|
class Deserializer(abc.ABC, Generic[T]):
|
|
"Parses a JSON value into a Python type."
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
"""
|
|
Creates auxiliary parsers that this parser is depending on.
|
|
|
|
:param context: A module context for evaluating types specified as a string.
|
|
"""
|
|
|
|
@abc.abstractmethod
|
|
def parse(self, data: JsonType) -> T:
|
|
"""
|
|
Parses a JSON value into a Python type.
|
|
|
|
:param data: The JSON value to de-serialize.
|
|
:returns: The Python object that the JSON value de-serializes to.
|
|
"""
|
|
|
|
|
|
class NoneDeserializer(Deserializer[None]):
|
|
"Parses JSON `null` values into Python `None`."
|
|
|
|
def parse(self, data: JsonType) -> None:
|
|
if data is not None:
|
|
raise JsonTypeError(
|
|
f"`None` type expects JSON `null` but instead received: {data}"
|
|
)
|
|
return None
|
|
|
|
|
|
class BoolDeserializer(Deserializer[bool]):
|
|
"Parses JSON `boolean` values into Python `bool` type."
|
|
|
|
def parse(self, data: JsonType) -> bool:
|
|
if not isinstance(data, bool):
|
|
raise JsonTypeError(
|
|
f"`bool` type expects JSON `boolean` data but instead received: {data}"
|
|
)
|
|
return bool(data)
|
|
|
|
|
|
class IntDeserializer(Deserializer[int]):
|
|
"Parses JSON `number` values into Python `int` type."
|
|
|
|
def parse(self, data: JsonType) -> int:
|
|
if not isinstance(data, int):
|
|
raise JsonTypeError(
|
|
f"`int` type expects integer data as JSON `number` but instead received: {data}"
|
|
)
|
|
return int(data)
|
|
|
|
|
|
class FloatDeserializer(Deserializer[float]):
|
|
"Parses JSON `number` values into Python `float` type."
|
|
|
|
def parse(self, data: JsonType) -> float:
|
|
if not isinstance(data, float) and not isinstance(data, int):
|
|
raise JsonTypeError(
|
|
f"`int` type expects data as JSON `number` but instead received: {data}"
|
|
)
|
|
return float(data)
|
|
|
|
|
|
class StringDeserializer(Deserializer[str]):
|
|
"Parses JSON `string` values into Python `str` type."
|
|
|
|
def parse(self, data: JsonType) -> str:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`str` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
return str(data)
|
|
|
|
|
|
class BytesDeserializer(Deserializer[bytes]):
|
|
"Parses JSON `string` values of Base64-encoded strings into Python `bytes` type."
|
|
|
|
def parse(self, data: JsonType) -> bytes:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`bytes` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
return base64.b64decode(data, validate=True)
|
|
|
|
|
|
class DateTimeDeserializer(Deserializer[datetime.datetime]):
|
|
"Parses JSON `string` values representing timestamps in ISO 8601 format to Python `datetime` with time zone."
|
|
|
|
def parse(self, data: JsonType) -> datetime.datetime:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`datetime` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
|
|
if data.endswith("Z"):
|
|
data = f"{data[:-1]}+00:00" # Python's isoformat() does not support military time zones like "Zulu" for UTC
|
|
timestamp = datetime.datetime.fromisoformat(data)
|
|
if timestamp.tzinfo is None:
|
|
raise JsonValueError(
|
|
f"timestamp lacks explicit time zone designator: {data}"
|
|
)
|
|
return timestamp
|
|
|
|
|
|
class DateDeserializer(Deserializer[datetime.date]):
|
|
"Parses JSON `string` values representing dates in ISO 8601 format to Python `date` type."
|
|
|
|
def parse(self, data: JsonType) -> datetime.date:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`date` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
|
|
return datetime.date.fromisoformat(data)
|
|
|
|
|
|
class TimeDeserializer(Deserializer[datetime.time]):
|
|
"Parses JSON `string` values representing time instances in ISO 8601 format to Python `time` type with time zone."
|
|
|
|
def parse(self, data: JsonType) -> datetime.time:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`time` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
|
|
return datetime.time.fromisoformat(data)
|
|
|
|
|
|
class UUIDDeserializer(Deserializer[uuid.UUID]):
|
|
"Parses JSON `string` values of UUID strings into Python `uuid.UUID` type."
|
|
|
|
def parse(self, data: JsonType) -> uuid.UUID:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`UUID` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
return uuid.UUID(data)
|
|
|
|
|
|
class IPv4Deserializer(Deserializer[ipaddress.IPv4Address]):
|
|
"Parses JSON `string` values of IPv4 address strings into Python `ipaddress.IPv4Address` type."
|
|
|
|
def parse(self, data: JsonType) -> ipaddress.IPv4Address:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`IPv4Address` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
return ipaddress.IPv4Address(data)
|
|
|
|
|
|
class IPv6Deserializer(Deserializer[ipaddress.IPv6Address]):
|
|
"Parses JSON `string` values of IPv6 address strings into Python `ipaddress.IPv6Address` type."
|
|
|
|
def parse(self, data: JsonType) -> ipaddress.IPv6Address:
|
|
if not isinstance(data, str):
|
|
raise JsonTypeError(
|
|
f"`IPv6Address` type expects JSON `string` data but instead received: {data}"
|
|
)
|
|
return ipaddress.IPv6Address(data)
|
|
|
|
|
|
class ListDeserializer(Deserializer[List[T]]):
|
|
"Recursively de-serializes a JSON array into a Python `list`."
|
|
|
|
item_type: Type[T]
|
|
item_parser: Deserializer
|
|
|
|
def __init__(self, item_type: Type[T]) -> None:
|
|
self.item_type = item_type
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
self.item_parser = _get_deserializer(self.item_type, context)
|
|
|
|
def parse(self, data: JsonType) -> List[T]:
|
|
if not isinstance(data, list):
|
|
type_name = python_type_to_str(self.item_type)
|
|
raise JsonTypeError(
|
|
f"type `List[{type_name}]` expects JSON `array` data but instead received: {data}"
|
|
)
|
|
|
|
return [self.item_parser.parse(item) for item in data]
|
|
|
|
|
|
class DictDeserializer(Deserializer[Dict[K, V]]):
|
|
"Recursively de-serializes a JSON object into a Python `dict`."
|
|
|
|
key_type: Type[K]
|
|
value_type: Type[V]
|
|
value_parser: Deserializer[V]
|
|
|
|
def __init__(self, key_type: Type[K], value_type: Type[V]) -> None:
|
|
self.key_type = key_type
|
|
self.value_type = value_type
|
|
self._check_key_type()
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
self.value_parser = _get_deserializer(self.value_type, context)
|
|
|
|
def _check_key_type(self) -> None:
|
|
if self.key_type is str:
|
|
return
|
|
|
|
if issubclass(self.key_type, enum.Enum):
|
|
value_types = enum_value_types(self.key_type)
|
|
if len(value_types) != 1:
|
|
raise JsonTypeError(
|
|
f"type `{self.container_type}` has invalid key type, "
|
|
f"enumerations must have a consistent member value type but several types found: {value_types}"
|
|
)
|
|
value_type = value_types.pop()
|
|
if value_type is not str:
|
|
f"`type `{self.container_type}` has invalid enumeration key type, expected `enum.Enum` with string values"
|
|
return
|
|
|
|
raise JsonTypeError(
|
|
f"`type `{self.container_type}` has invalid key type, expected `str` or `enum.Enum` with string values"
|
|
)
|
|
|
|
@property
|
|
def container_type(self) -> str:
|
|
key_type_name = python_type_to_str(self.key_type)
|
|
value_type_name = python_type_to_str(self.value_type)
|
|
return f"Dict[{key_type_name}, {value_type_name}]"
|
|
|
|
def parse(self, data: JsonType) -> Dict[K, V]:
|
|
if not isinstance(data, dict):
|
|
raise JsonTypeError(
|
|
f"`type `{self.container_type}` expects JSON `object` data but instead received: {data}"
|
|
)
|
|
|
|
return dict(
|
|
(self.key_type(key), self.value_parser.parse(value)) # type: ignore[call-arg]
|
|
for key, value in data.items()
|
|
)
|
|
|
|
|
|
class SetDeserializer(Deserializer[Set[T]]):
|
|
"Recursively de-serializes a JSON list into a Python `set`."
|
|
|
|
member_type: Type[T]
|
|
member_parser: Deserializer
|
|
|
|
def __init__(self, member_type: Type[T]) -> None:
|
|
self.member_type = member_type
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
self.member_parser = _get_deserializer(self.member_type, context)
|
|
|
|
def parse(self, data: JsonType) -> Set[T]:
|
|
if not isinstance(data, list):
|
|
type_name = python_type_to_str(self.member_type)
|
|
raise JsonTypeError(
|
|
f"type `Set[{type_name}]` expects JSON `array` data but instead received: {data}"
|
|
)
|
|
|
|
return set(self.member_parser.parse(item) for item in data)
|
|
|
|
|
|
class TupleDeserializer(Deserializer[Tuple[Any, ...]]):
|
|
"Recursively de-serializes a JSON list into a Python `tuple`."
|
|
|
|
item_types: Tuple[Type[Any], ...]
|
|
item_parsers: Tuple[Deserializer[Any], ...]
|
|
|
|
def __init__(self, item_types: Tuple[Type[Any], ...]) -> None:
|
|
self.item_types = item_types
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
self.item_parsers = tuple(
|
|
_get_deserializer(item_type, context) for item_type in self.item_types
|
|
)
|
|
|
|
@property
|
|
def container_type(self) -> str:
|
|
type_names = ", ".join(
|
|
python_type_to_str(item_type) for item_type in self.item_types
|
|
)
|
|
return f"Tuple[{type_names}]"
|
|
|
|
def parse(self, data: JsonType) -> Tuple[Any, ...]:
|
|
if not isinstance(data, list) or len(data) != len(self.item_parsers):
|
|
if not isinstance(data, list):
|
|
raise JsonTypeError(
|
|
f"type `{self.container_type}` expects JSON `array` data but instead received: {data}"
|
|
)
|
|
else:
|
|
count = len(self.item_parsers)
|
|
raise JsonValueError(
|
|
f"type `{self.container_type}` expects a JSON `array` of length {count} but received length {len(data)}"
|
|
)
|
|
|
|
return tuple(
|
|
item_parser.parse(item)
|
|
for item_parser, item in zip(self.item_parsers, data)
|
|
)
|
|
|
|
|
|
class UnionDeserializer(Deserializer):
|
|
"De-serializes a JSON value (of any type) into a Python union type."
|
|
|
|
member_types: Tuple[type, ...]
|
|
member_parsers: Tuple[Deserializer, ...]
|
|
|
|
def __init__(self, member_types: Tuple[type, ...]) -> None:
|
|
self.member_types = member_types
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
self.member_parsers = tuple(
|
|
_get_deserializer(member_type, context) for member_type in self.member_types
|
|
)
|
|
|
|
def parse(self, data: JsonType) -> Any:
|
|
for member_parser in self.member_parsers:
|
|
# iterate over potential types of discriminated union
|
|
try:
|
|
return member_parser.parse(data)
|
|
except (JsonKeyError, JsonTypeError):
|
|
# indicates a required field is missing from JSON dict -OR- the data cannot be cast to the expected type,
|
|
# i.e. we don't have the type that we are looking for
|
|
continue
|
|
|
|
type_names = ", ".join(
|
|
python_type_to_str(member_type) for member_type in self.member_types
|
|
)
|
|
raise JsonKeyError(
|
|
f"type `Union[{type_names}]` could not be instantiated from: {data}"
|
|
)
|
|
|
|
|
|
def get_literal_properties(typ: type) -> Set[str]:
|
|
"Returns the names of all properties in a class that are of a literal type."
|
|
|
|
return set(
|
|
property_name
|
|
for property_name, property_type in get_class_properties(typ)
|
|
if is_type_literal(property_type)
|
|
)
|
|
|
|
|
|
def get_discriminating_properties(types: Tuple[type, ...]) -> Set[str]:
|
|
"Returns a set of properties with literal type that are common across all specified classes."
|
|
|
|
if not types or not all(isinstance(typ, type) for typ in types):
|
|
return set()
|
|
|
|
props = get_literal_properties(types[0])
|
|
for typ in types[1:]:
|
|
props = props & get_literal_properties(typ)
|
|
|
|
return props
|
|
|
|
|
|
class TaggedUnionDeserializer(Deserializer):
|
|
"De-serializes a JSON value with one or more disambiguating properties into a Python union type."
|
|
|
|
member_types: Tuple[type, ...]
|
|
disambiguating_properties: Set[str]
|
|
member_parsers: Dict[Tuple[str, Any], Deserializer]
|
|
|
|
def __init__(self, member_types: Tuple[type, ...]) -> None:
|
|
self.member_types = member_types
|
|
self.disambiguating_properties = get_discriminating_properties(member_types)
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
self.member_parsers = {}
|
|
for member_type in self.member_types:
|
|
for property_name in self.disambiguating_properties:
|
|
literal_type = get_class_property(member_type, property_name)
|
|
if not literal_type:
|
|
continue
|
|
|
|
for literal_value in unwrap_literal_values(literal_type):
|
|
tpl = (property_name, literal_value)
|
|
if tpl in self.member_parsers:
|
|
raise JsonTypeError(
|
|
f"disambiguating property `{property_name}` in type `{self.union_type}` has a duplicate value: {literal_value}"
|
|
)
|
|
|
|
self.member_parsers[tpl] = _get_deserializer(member_type, context)
|
|
|
|
@property
|
|
def union_type(self) -> str:
|
|
type_names = ", ".join(
|
|
python_type_to_str(member_type) for member_type in self.member_types
|
|
)
|
|
return f"Union[{type_names}]"
|
|
|
|
def parse(self, data: JsonType) -> Any:
|
|
if not isinstance(data, dict):
|
|
raise JsonTypeError(
|
|
f"tagged union type `{self.union_type}` expects JSON `object` data but instead received: {data}"
|
|
)
|
|
|
|
for property_name in self.disambiguating_properties:
|
|
disambiguating_value = data.get(property_name)
|
|
if disambiguating_value is None:
|
|
continue
|
|
|
|
member_parser = self.member_parsers.get(
|
|
(property_name, disambiguating_value)
|
|
)
|
|
if member_parser is None:
|
|
raise JsonTypeError(
|
|
f"disambiguating property value is invalid for tagged union type `{self.union_type}`: {data}"
|
|
)
|
|
|
|
return member_parser.parse(data)
|
|
|
|
raise JsonTypeError(
|
|
f"disambiguating property value is missing for tagged union type `{self.union_type}`: {data}"
|
|
)
|
|
|
|
|
|
class LiteralDeserializer(Deserializer):
|
|
"De-serializes a JSON value into a Python literal type."
|
|
|
|
values: Tuple[Any, ...]
|
|
parser: Deserializer
|
|
|
|
def __init__(self, values: Tuple[Any, ...]) -> None:
|
|
self.values = values
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
literal_type_tuple = tuple(type(value) for value in self.values)
|
|
literal_type_set = set(literal_type_tuple)
|
|
if len(literal_type_set) != 1:
|
|
value_names = ", ".join(repr(value) for value in self.values)
|
|
raise TypeError(
|
|
f"type `Literal[{value_names}]` expects consistent literal value types but got: {literal_type_tuple}"
|
|
)
|
|
|
|
literal_type = literal_type_set.pop()
|
|
self.parser = _get_deserializer(literal_type, context)
|
|
|
|
def parse(self, data: JsonType) -> Any:
|
|
value = self.parser.parse(data)
|
|
if value not in self.values:
|
|
value_names = ", ".join(repr(value) for value in self.values)
|
|
raise JsonTypeError(
|
|
f"type `Literal[{value_names}]` could not be instantiated from: {data}"
|
|
)
|
|
return value
|
|
|
|
|
|
class EnumDeserializer(Deserializer[E]):
|
|
"Returns an enumeration instance based on the enumeration value read from a JSON value."
|
|
|
|
enum_type: Type[E]
|
|
|
|
def __init__(self, enum_type: Type[E]) -> None:
|
|
self.enum_type = enum_type
|
|
|
|
def parse(self, data: JsonType) -> E:
|
|
return self.enum_type(data)
|
|
|
|
|
|
class CustomDeserializer(Deserializer[T]):
|
|
"Uses the `from_json` class method in class to de-serialize the object from JSON."
|
|
|
|
converter: Callable[[JsonType], T]
|
|
|
|
def __init__(self, converter: Callable[[JsonType], T]) -> None:
|
|
self.converter = converter
|
|
|
|
def parse(self, data: JsonType) -> T:
|
|
return self.converter(data)
|
|
|
|
|
|
class FieldDeserializer(abc.ABC, Generic[T, R]):
|
|
"""
|
|
Deserializes a JSON property into a Python object field.
|
|
|
|
:param property_name: The name of the JSON property to read from a JSON `object`.
|
|
:param field_name: The name of the field in a Python class to write data to.
|
|
:param parser: A compatible deserializer that can handle the field's type.
|
|
"""
|
|
|
|
property_name: str
|
|
field_name: str
|
|
parser: Deserializer[T]
|
|
|
|
def __init__(
|
|
self, property_name: str, field_name: str, parser: Deserializer[T]
|
|
) -> None:
|
|
self.property_name = property_name
|
|
self.field_name = field_name
|
|
self.parser = parser
|
|
|
|
@abc.abstractmethod
|
|
def parse_field(self, data: Dict[str, JsonType]) -> R: ...
|
|
|
|
|
|
class RequiredFieldDeserializer(FieldDeserializer[T, T]):
|
|
"Deserializes a JSON property into a mandatory Python object field."
|
|
|
|
def parse_field(self, data: Dict[str, JsonType]) -> T:
|
|
if self.property_name not in data:
|
|
raise JsonKeyError(
|
|
f"missing required property `{self.property_name}` from JSON object: {data}"
|
|
)
|
|
|
|
return self.parser.parse(data[self.property_name])
|
|
|
|
|
|
class OptionalFieldDeserializer(FieldDeserializer[T, Optional[T]]):
|
|
"Deserializes a JSON property into an optional Python object field with a default value of `None`."
|
|
|
|
def parse_field(self, data: Dict[str, JsonType]) -> Optional[T]:
|
|
value = data.get(self.property_name)
|
|
if value is not None:
|
|
return self.parser.parse(value)
|
|
else:
|
|
return None
|
|
|
|
|
|
class DefaultFieldDeserializer(FieldDeserializer[T, T]):
|
|
"Deserializes a JSON property into a Python object field with an explicit default value."
|
|
|
|
default_value: T
|
|
|
|
def __init__(
|
|
self,
|
|
property_name: str,
|
|
field_name: str,
|
|
parser: Deserializer,
|
|
default_value: T,
|
|
) -> None:
|
|
super().__init__(property_name, field_name, parser)
|
|
self.default_value = default_value
|
|
|
|
def parse_field(self, data: Dict[str, JsonType]) -> T:
|
|
value = data.get(self.property_name)
|
|
if value is not None:
|
|
return self.parser.parse(value)
|
|
else:
|
|
return self.default_value
|
|
|
|
|
|
class DefaultFactoryFieldDeserializer(FieldDeserializer[T, T]):
|
|
"Deserializes a JSON property into an optional Python object field with an explicit default value factory."
|
|
|
|
default_factory: Callable[[], T]
|
|
|
|
def __init__(
|
|
self,
|
|
property_name: str,
|
|
field_name: str,
|
|
parser: Deserializer[T],
|
|
default_factory: Callable[[], T],
|
|
) -> None:
|
|
super().__init__(property_name, field_name, parser)
|
|
self.default_factory = default_factory
|
|
|
|
def parse_field(self, data: Dict[str, JsonType]) -> T:
|
|
value = data.get(self.property_name)
|
|
if value is not None:
|
|
return self.parser.parse(value)
|
|
else:
|
|
return self.default_factory()
|
|
|
|
|
|
class ClassDeserializer(Deserializer[T]):
|
|
"Base class for de-serializing class-like types such as data classes, named tuples and regular classes."
|
|
|
|
class_type: type
|
|
property_parsers: List[FieldDeserializer]
|
|
property_fields: Set[str]
|
|
|
|
def __init__(self, class_type: Type[T]) -> None:
|
|
self.class_type = class_type
|
|
|
|
def assign(self, property_parsers: List[FieldDeserializer]) -> None:
|
|
self.property_parsers = property_parsers
|
|
self.property_fields = set(
|
|
property_parser.property_name for property_parser in property_parsers
|
|
)
|
|
|
|
def parse(self, data: JsonType) -> T:
|
|
if not isinstance(data, dict):
|
|
type_name = python_type_to_str(self.class_type)
|
|
raise JsonTypeError(
|
|
f"`type `{type_name}` expects JSON `object` data but instead received: {data}"
|
|
)
|
|
|
|
object_data: Dict[str, JsonType] = typing.cast(Dict[str, JsonType], data)
|
|
|
|
field_values = {}
|
|
for property_parser in self.property_parsers:
|
|
field_values[property_parser.field_name] = property_parser.parse_field(
|
|
object_data
|
|
)
|
|
|
|
if not self.property_fields.issuperset(object_data):
|
|
unassigned_names = [
|
|
name for name in object_data if name not in self.property_fields
|
|
]
|
|
raise JsonKeyError(
|
|
f"unrecognized fields in JSON object: {unassigned_names}"
|
|
)
|
|
|
|
return self.create(**field_values)
|
|
|
|
def create(self, **field_values: Any) -> T:
|
|
"Instantiates an object with a collection of property values."
|
|
|
|
obj: T = create_object(self.class_type)
|
|
|
|
# use `setattr` on newly created object instance
|
|
for field_name, field_value in field_values.items():
|
|
setattr(obj, field_name, field_value)
|
|
return obj
|
|
|
|
|
|
class NamedTupleDeserializer(ClassDeserializer[NamedTuple]):
|
|
"De-serializes a named tuple from a JSON `object`."
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
property_parsers: List[FieldDeserializer] = [
|
|
RequiredFieldDeserializer(
|
|
field_name, field_name, _get_deserializer(field_type, context)
|
|
)
|
|
for field_name, field_type in get_resolved_hints(self.class_type).items()
|
|
]
|
|
super().assign(property_parsers)
|
|
|
|
def create(self, **field_values: Any) -> NamedTuple:
|
|
return self.class_type(**field_values)
|
|
|
|
|
|
class DataclassDeserializer(ClassDeserializer[T]):
|
|
"De-serializes a data class from a JSON `object`."
|
|
|
|
def __init__(self, class_type: Type[T]) -> None:
|
|
if not dataclasses.is_dataclass(class_type):
|
|
raise TypeError("expected: data-class type")
|
|
super().__init__(class_type) # type: ignore[arg-type]
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
property_parsers: List[FieldDeserializer] = []
|
|
resolved_hints = get_resolved_hints(self.class_type)
|
|
for field in dataclasses.fields(self.class_type):
|
|
field_type = resolved_hints[field.name]
|
|
property_name = python_field_to_json_property(field.name, field_type)
|
|
|
|
is_optional = is_type_optional(field_type)
|
|
has_default = field.default is not dataclasses.MISSING
|
|
has_default_factory = field.default_factory is not dataclasses.MISSING
|
|
|
|
if is_optional:
|
|
required_type: Type[T] = unwrap_optional_type(field_type)
|
|
else:
|
|
required_type = field_type
|
|
|
|
parser = _get_deserializer(required_type, context)
|
|
|
|
if has_default:
|
|
field_parser: FieldDeserializer = DefaultFieldDeserializer(
|
|
property_name, field.name, parser, field.default
|
|
)
|
|
elif has_default_factory:
|
|
default_factory = typing.cast(Callable[[], Any], field.default_factory)
|
|
field_parser = DefaultFactoryFieldDeserializer(
|
|
property_name, field.name, parser, default_factory
|
|
)
|
|
elif is_optional:
|
|
field_parser = OptionalFieldDeserializer(
|
|
property_name, field.name, parser
|
|
)
|
|
else:
|
|
field_parser = RequiredFieldDeserializer(
|
|
property_name, field.name, parser
|
|
)
|
|
|
|
property_parsers.append(field_parser)
|
|
|
|
super().assign(property_parsers)
|
|
|
|
|
|
class FrozenDataclassDeserializer(DataclassDeserializer[T]):
|
|
"De-serializes a frozen data class from a JSON `object`."
|
|
|
|
def create(self, **field_values: Any) -> T:
|
|
"Instantiates an object with a collection of property values."
|
|
|
|
# create object instance without calling `__init__`
|
|
obj: T = create_object(self.class_type)
|
|
|
|
# can't use `setattr` on frozen dataclasses, pass member variable values to `__init__`
|
|
obj.__init__(**field_values) # type: ignore
|
|
return obj
|
|
|
|
|
|
class TypedClassDeserializer(ClassDeserializer[T]):
|
|
"De-serializes a class with type annotations from a JSON `object` by iterating over class properties."
|
|
|
|
def build(self, context: Optional[ModuleType]) -> None:
|
|
property_parsers: List[FieldDeserializer] = []
|
|
for field_name, field_type in get_resolved_hints(self.class_type).items():
|
|
property_name = python_field_to_json_property(field_name, field_type)
|
|
|
|
is_optional = is_type_optional(field_type)
|
|
|
|
if is_optional:
|
|
required_type: Type[T] = unwrap_optional_type(field_type)
|
|
else:
|
|
required_type = field_type
|
|
|
|
parser = _get_deserializer(required_type, context)
|
|
|
|
if is_optional:
|
|
field_parser: FieldDeserializer = OptionalFieldDeserializer(
|
|
property_name, field_name, parser
|
|
)
|
|
else:
|
|
field_parser = RequiredFieldDeserializer(
|
|
property_name, field_name, parser
|
|
)
|
|
|
|
property_parsers.append(field_parser)
|
|
|
|
super().assign(property_parsers)
|
|
|
|
|
|
def create_deserializer(
|
|
typ: TypeLike, context: Optional[ModuleType] = None
|
|
) -> Deserializer:
|
|
"""
|
|
Creates a de-serializer engine to produce a Python object from an object obtained from a JSON string.
|
|
|
|
When de-serializing a JSON object into a Python object, the following transformations are applied:
|
|
|
|
* Fundamental types are parsed as `bool`, `int`, `float` or `str`.
|
|
* Date and time types are parsed from the ISO 8601 format with time zone into the corresponding Python type
|
|
`datetime`, `date` or `time`.
|
|
* Byte arrays are read from a string with Base64 encoding into a `bytes` instance.
|
|
* UUIDs are extracted from a UUID string compliant with RFC 4122 into a `uuid.UUID` instance.
|
|
* Enumerations are instantiated with a lookup on enumeration value.
|
|
* Containers (e.g. `list`, `dict`, `set`, `tuple`) are parsed recursively.
|
|
* Complex objects with properties (including data class types) are populated from dictionaries of key-value pairs
|
|
using reflection (enumerating type annotations).
|
|
|
|
:raises TypeError: A de-serializer engine cannot be constructed for the input type.
|
|
"""
|
|
|
|
if context is None:
|
|
if isinstance(typ, type):
|
|
context = sys.modules[typ.__module__]
|
|
|
|
return _get_deserializer(typ, context)
|
|
|
|
|
|
_CACHE: Dict[Tuple[str, str], Deserializer] = {}
|
|
|
|
|
|
def _get_deserializer(typ: TypeLike, context: Optional[ModuleType]) -> Deserializer:
|
|
"Creates or re-uses a de-serializer engine to parse an object obtained from a JSON string."
|
|
|
|
cache_key = None
|
|
|
|
if isinstance(typ, (str, typing.ForwardRef)):
|
|
if context is None:
|
|
raise TypeError(f"missing context for evaluating type: {typ}")
|
|
|
|
if isinstance(typ, str):
|
|
if hasattr(context, typ):
|
|
cache_key = (context.__name__, typ)
|
|
elif isinstance(typ, typing.ForwardRef):
|
|
if hasattr(context, typ.__forward_arg__):
|
|
cache_key = (context.__name__, typ.__forward_arg__)
|
|
|
|
typ = evaluate_type(typ, context)
|
|
|
|
typ = unwrap_annotated_type(typ) if is_type_annotated(typ) else typ
|
|
|
|
if isinstance(typ, type) and typing.get_origin(typ) is None:
|
|
cache_key = (typ.__module__, typ.__name__)
|
|
|
|
if cache_key is not None:
|
|
deserializer = _CACHE.get(cache_key)
|
|
if deserializer is None:
|
|
deserializer = _create_deserializer(typ)
|
|
|
|
# store de-serializer immediately in cache to avoid stack overflow for recursive types
|
|
_CACHE[cache_key] = deserializer
|
|
|
|
if isinstance(typ, type):
|
|
# use type's own module as context for evaluating member types
|
|
context = sys.modules[typ.__module__]
|
|
|
|
# create any de-serializers this de-serializer is depending on
|
|
deserializer.build(context)
|
|
else:
|
|
# special forms are not always hashable, create a new de-serializer every time
|
|
deserializer = _create_deserializer(typ)
|
|
deserializer.build(context)
|
|
|
|
return deserializer
|
|
|
|
|
|
def _create_deserializer(typ: TypeLike) -> Deserializer:
|
|
"Creates a de-serializer engine to parse an object obtained from a JSON string."
|
|
|
|
# check for well-known types
|
|
if typ is type(None):
|
|
return NoneDeserializer()
|
|
elif typ is bool:
|
|
return BoolDeserializer()
|
|
elif typ is int:
|
|
return IntDeserializer()
|
|
elif typ is float:
|
|
return FloatDeserializer()
|
|
elif typ is str:
|
|
return StringDeserializer()
|
|
elif typ is bytes:
|
|
return BytesDeserializer()
|
|
elif typ is datetime.datetime:
|
|
return DateTimeDeserializer()
|
|
elif typ is datetime.date:
|
|
return DateDeserializer()
|
|
elif typ is datetime.time:
|
|
return TimeDeserializer()
|
|
elif typ is uuid.UUID:
|
|
return UUIDDeserializer()
|
|
elif typ is ipaddress.IPv4Address:
|
|
return IPv4Deserializer()
|
|
elif typ is ipaddress.IPv6Address:
|
|
return IPv6Deserializer()
|
|
|
|
# dynamically-typed collection types
|
|
if typ is list:
|
|
raise TypeError("explicit item type required: use `List[T]` instead of `list`")
|
|
if typ is dict:
|
|
raise TypeError(
|
|
"explicit key and value types required: use `Dict[K, V]` instead of `dict`"
|
|
)
|
|
if typ is set:
|
|
raise TypeError("explicit member type required: use `Set[T]` instead of `set`")
|
|
if typ is tuple:
|
|
raise TypeError(
|
|
"explicit item type list required: use `Tuple[T, ...]` instead of `tuple`"
|
|
)
|
|
|
|
# generic types (e.g. list, dict, set, etc.)
|
|
origin_type = typing.get_origin(typ)
|
|
if origin_type is list:
|
|
(list_item_type,) = typing.get_args(typ) # unpack single tuple element
|
|
return ListDeserializer(list_item_type)
|
|
elif origin_type is dict:
|
|
key_type, value_type = typing.get_args(typ)
|
|
return DictDeserializer(key_type, value_type)
|
|
elif origin_type is set:
|
|
(set_member_type,) = typing.get_args(typ) # unpack single tuple element
|
|
return SetDeserializer(set_member_type)
|
|
elif origin_type is tuple:
|
|
return TupleDeserializer(typing.get_args(typ))
|
|
elif origin_type is Union:
|
|
union_args = typing.get_args(typ)
|
|
if get_discriminating_properties(union_args):
|
|
return TaggedUnionDeserializer(union_args)
|
|
else:
|
|
return UnionDeserializer(union_args)
|
|
elif origin_type is Literal:
|
|
return LiteralDeserializer(typing.get_args(typ))
|
|
|
|
if not inspect.isclass(typ):
|
|
if is_dataclass_instance(typ):
|
|
raise TypeError(f"dataclass type expected but got instance: {typ}")
|
|
else:
|
|
raise TypeError(f"unable to de-serialize unrecognized type: {typ}")
|
|
|
|
if issubclass(typ, enum.Enum):
|
|
return EnumDeserializer(typ)
|
|
|
|
if is_named_tuple_type(typ):
|
|
return NamedTupleDeserializer(typ)
|
|
|
|
# check if object has custom serialization method
|
|
convert_func = getattr(typ, "from_json", None)
|
|
if callable(convert_func):
|
|
return CustomDeserializer(convert_func)
|
|
|
|
if is_dataclass_type(typ):
|
|
dataclass_params = getattr(typ, "__dataclass_params__", None)
|
|
if dataclass_params is not None and dataclass_params.frozen:
|
|
return FrozenDataclassDeserializer(typ)
|
|
else:
|
|
return DataclassDeserializer(typ)
|
|
|
|
return TypedClassDeserializer(typ)
|