Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion csp/impl/struct.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,18 @@ def serializer(val, handler):


class Struct(_csptypesimpl.PyStruct, metaclass=StructMeta):
@classmethod
def type_adapter(cls):
internal_type_adapter = getattr(cls, "_pydantic_type_adapter", None)
if internal_type_adapter:
return internal_type_adapter

# Late import to avoid autogen issues
from pydantic import TypeAdapter

cls._pydantic_type_adapter = TypeAdapter(cls)
return cls._pydantic_type_adapter

@classmethod
def metadata(cls, typed=False):
if typed:
Expand Down Expand Up @@ -235,7 +247,9 @@ def _obj_from_python(cls, json, obj_type):
return obj_type(json)

@classmethod
def from_dict(cls, json: dict):
def from_dict(cls, json: dict, use_pydantic: bool = False):
if use_pydantic:
return cls.type_adapter().validate_python(json)
return cls._obj_from_python(json, cls)

def to_dict_depr(self):
Expand Down
27 changes: 27 additions & 0 deletions csp/impl/types/typing_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,33 @@ class FastList(typing.List, typing.Generic[T]): # Need to inherit from Generic[
def __init__(self):
raise NotImplementedError("Can not init FastList class")

@classmethod
def __get_pydantic_core_schema__(cls, source_type, handler):
from pydantic_core import core_schema

# Late import to not interfere with autogen
args = typing.get_args(source_type)
if args:
inner_type = args[0]
list_schema = handler.generate_schema(typing.List[inner_type])
else:
list_schema = handler.generate_schema(typing.List)

def create_instance(raw_data, validator):
if isinstance(raw_data, FastList):
return raw_data
return validator(raw_data) # just return a list

return core_schema.no_info_wrap_validator_function(
function=create_instance,
schema=list_schema,
serialization=core_schema.plain_serializer_function_ser_schema(
lambda val: list(v for v in val),
return_schema=list_schema,
when_used="json",
),
)


class CspTypingUtils39:
_ORIGIN_COMPAT_MAP = {list: typing.List, set: typing.Set, dict: typing.Dict, tuple: typing.Tuple}
Expand Down
Loading
Loading