""" """
import importlib
import json
from typing import Any
[docs]
class TypeDescriptor:
[docs]
def __init__(self, type_string: str, extra_init_data: dict):
self.type_string = type_string
self.extra_init_data = extra_init_data or {}
self.is_json: bool = False
self.type: str = self._extract_type()
def _extract_type(self) -> str:
if self.type_string.startswith("<class '"):
return self.type_string[8:-2]
# Handle Annotated case of FastAPI annotation
if "typing.Optional" in self.type_string:
annotated = _extract_type_of(self.type_string)
if "typing.Annotated" in annotated:
# Extract content between the first square brackets after Annotated
match = _extract_type_of(annotated)
if match:
# Split by comma and get first part, then strip whitespace
parts = [p.strip() for p in match.split(",")]
if len(parts) > 1:
self.is_json = parts[1].lower() == "json"
return parts[0] if parts else ""
else:
# Handle simple Optional case
return annotated
return _extract_type_of(self.type_string)
[docs]
def instantiate(self, payload: Any, is_json: bool = False) -> Any:
return _instantiate(
(
self.type
if "typing.Optional" in self.type_string or self.type_string.startswith("<class ")
else self.type_string
),
payload,
self.is_json or is_json,
extra_init_data=self.extra_init_data,
)
def __repr__(self) -> str:
return f"TypeDescriptor(type='{self.type}', is_json={self.is_json})"
def _extract_type_of(input_string: str) -> str:
stack = []
start_index = -1
end_index = -1
for i, char in enumerate(input_string):
if i > 0: # this assumes paramerized type, that must be at lest 1 char long
if char == "[":
if start_index == -1:
start_index = i
stack.append(char)
elif char == "]":
if len(stack) == 1:
end_index = i
break
if len(stack) > 0:
stack.pop()
if start_index != -1 and end_index != -1:
return input_string[start_index + 1 : end_index].strip()
else:
return input_string.strip() if start_index == 1 and end_index == -1 else ""
def _instantiate(
type_string: str, payload: Any, is_json: bool = False, extra_init_data: dict = None
) -> Any:
"""
Dynamically imports and instantiates a class from a TypeDescriptor's type_string attribute.
Args:
type_string: TypeDescriptor instance with the class path
payload: Argument to pass to the constructor
Returns:
Instance of the target class initialized with payload
Raises:
ImportError: If the module/class cannot be imported
TypeError: If instantiation fails
"""
# Handle JSON payload first
json_payload = json.loads(payload) if is_json and isinstance(payload, str) else None
# Handle basic types
if type_string == "str":
return str(payload)
if type_string == "int":
return int(payload)
if type_string == "float":
return float(payload)
if type_string == "bool":
return bool(payload)
# Handle typing module types
if type_string.startswith("typing.") or type_string.startswith("list"):
if (
type_string.startswith("typing.List")
or type_string.startswith("typing.Sequence")
or type_string.startswith("list")
):
if json_payload is not None:
return json_payload
inner_type = _extract_type_of(type_string)
if inner_type:
return [
_instantiate(inner_type, p, extra_init_data=extra_init_data)
for p in (payload if isinstance(payload, list) else payload.split(","))
]
return payload if isinstance(payload, list) else payload.split(",")
if type_string.startswith("typing.Dict") or type_string.startswith("typing.Mapping"):
if json_payload is not None:
return json_payload
return dict(payload)
if type_string.startswith("typing.Set"):
if json_payload is not None:
return set(json_payload)
_set_type = _extract_type_of(type_string)
return set(
payload
if isinstance(payload, list)
else [_instantiate(_set_type, p) for p in payload.split(",")]
)
if type_string.startswith("typing.Tuple"):
if json_payload is not None:
return tuple(json_payload)
_tuple_type = _extract_type_of(type_string)
return tuple(
payload
if isinstance(payload, list)
else [_instantiate(_tuple_type, p) for p in payload.split(",")]
)
if type_string.startswith("typing.Optional"):
inner_type = _extract_type_of(type_string)
if inner_type:
return _instantiate(inner_type, payload, is_json, extra_init_data)
return payload
# Handle regular class instantiation
parts = type_string.split(".")
module_path = ".".join(parts[:-1])
class_name = parts[-1]
try:
# Import the module
module = importlib.import_module(module_path)
# Get the class
cls = getattr(module, class_name)
if isinstance(payload, cls):
return payload
# Instantiate with payload
if json_payload is not None:
_args = merge_extra_init_data(cls, json_payload, extra_init_data or {})
if hasattr(cls, "from_dict"):
return cls.from_dict(_args)
elif hasattr(cls, "create"):
return cls.create(**_args)
return cls(**_args)
if isinstance(payload, dict):
_args = merge_extra_init_data(cls, payload, extra_init_data or {})
if hasattr(cls, "from_dict"):
return cls.from_dict(_args)
elif hasattr(cls, "create"):
return cls.create(**_args)
return cls(**_args)
return cls(payload)
except ImportError as e:
raise ImportError(f"Could not import module '{module_path}'") from e
except AttributeError as e:
raise ImportError(f"Module '{module_path}' has no class '{class_name}'") from e
except TypeError as e:
raise TypeError(f"Failed to instantiate {type_string} with payload {payload}") from e