Separate decoding of complex values into specific method

pydantic_settings/sources.py

@@ -84,9 +84,22 @@ def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, val
             The prepared value.
         """
         if value is not None and (self.field_is_complex(field) or value_is_complex):
-            return json.loads(value)
+            return self.decode_complex_value(field_name, field, value)
         return value
 
+    def decode_complex_value(self, field_name: str, field: FieldInfo, value: Any) -> Any:
+        """
+        Decode the value for a complex field
+
+        Args:
+            value: The value of the field which is to be decoded
+            field: The field which is being decoded
+
+        Returns:
+            The decoded value for further preparation
+        """
+        return json.loads(value)
+
     @abstractmethod
     def __call__(self) -> dict[str, Any]:
         pass
@@ -414,7 +427,7 @@ def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, val
             else:
                 # field is complex and there's a value, decode that as JSON, then add explode_env_vars
                 try:
-                    value = json.loads(value)
+                    value = self.decode_complex_value(field_name, field, value)
                 except ValueError as e:
                     if not allow_parse_failure:
                         raise e
@@ -516,7 +529,7 @@ def explode_env_vars(self, field_name: str, field: FieldInfo, env_vars: Mapping[
                 is_complex, allow_json_failure = self._field_is_complex(target_field)
                 if is_complex:
                     try:
-                        env_val = json.loads(env_val)
+                        env_val = self.decode_complex_value(last_key, target_field, env_val)
                     except ValueError as e:
                         if not allow_json_failure:
                             raise e