fixup!
This commit is contained in:
+20
-16
@@ -8,7 +8,7 @@ import re
|
|||||||
import time
|
import time
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Any, Generic, Literal, Optional, TypeVar
|
from typing import Any, Generic, Literal, Optional, TypeVar, Union
|
||||||
|
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
from pydantic import BaseModel, ValidationError, root_validator
|
from pydantic import BaseModel, ValidationError, root_validator
|
||||||
@@ -600,8 +600,8 @@ def model_to_dict(model: BaseModel) -> dict:
|
|||||||
private fields starting with _ are ignored
|
private fields starting with _ are ignored
|
||||||
:param model: Pydantic model
|
:param model: Pydantic model
|
||||||
"""
|
"""
|
||||||
_dict = model.dict()
|
_dict = {}
|
||||||
for key, value in _dict.items():
|
for key, value in model.dict().items():
|
||||||
if key.startswith("_"):
|
if key.startswith("_"):
|
||||||
continue
|
continue
|
||||||
type_ = model.__fields__[key].type_
|
type_ = model.__fields__[key].type_
|
||||||
@@ -610,6 +610,20 @@ def model_to_dict(model: BaseModel) -> dict:
|
|||||||
return _dict
|
return _dict
|
||||||
|
|
||||||
|
|
||||||
|
def dict_to_submodel(model: type[TModel], value: Union[dict, str]) -> Optional[TModel]:
|
||||||
|
if isinstance(value, str):
|
||||||
|
if value == "null":
|
||||||
|
return None
|
||||||
|
_subdict = json.loads(value)
|
||||||
|
elif isinstance(value, dict):
|
||||||
|
_subdict = value
|
||||||
|
else:
|
||||||
|
logger.warning(f"Expected str or dict, got {type(value)}")
|
||||||
|
return None
|
||||||
|
# recursively convert nested models
|
||||||
|
return dict_to_model(_subdict, model)
|
||||||
|
|
||||||
|
|
||||||
def dict_to_model(_row: dict, model: type[TModel]) -> TModel:
|
def dict_to_model(_row: dict, model: type[TModel]) -> TModel:
|
||||||
"""
|
"""
|
||||||
Convert a dictionary with JSON-encoded nested models to a Pydantic model
|
Convert a dictionary with JSON-encoded nested models to a Pydantic model
|
||||||
@@ -624,19 +638,9 @@ def dict_to_model(_row: dict, model: type[TModel]) -> TModel:
|
|||||||
if not value:
|
if not value:
|
||||||
continue
|
continue
|
||||||
type_ = model.__fields__[key].type_
|
type_ = model.__fields__[key].type_
|
||||||
if issubclass(type_, BaseModel) and value is not None:
|
if issubclass(type_, BaseModel):
|
||||||
if isinstance(value, str):
|
_dict[key] = dict_to_submodel(type_, value)
|
||||||
if value == "null":
|
|
||||||
_dict[key] = None
|
|
||||||
continue
|
|
||||||
_subdict = json.loads(value)
|
|
||||||
elif isinstance(value, dict):
|
|
||||||
_subdict = value
|
|
||||||
else:
|
|
||||||
logger.warning(f"Expected str or dict, got {type(value)}")
|
|
||||||
continue
|
|
||||||
# recursively convert nested models
|
|
||||||
_dict[key] = dict_to_model(_subdict, type_)
|
|
||||||
continue
|
continue
|
||||||
_dict[key] = value
|
_dict[key] = value
|
||||||
|
continue
|
||||||
return model.construct(**_dict)
|
return model.construct(**_dict)
|
||||||
|
|||||||
Reference in New Issue
Block a user