-
Notifications
You must be signed in to change notification settings - Fork 115
Description
What are you really trying to do?
I'm trying to use Temporal's value_to_type
function to convert JSON payloads to dataclasses that contain dictionary fields with custom string types as keys (either string subclasses or NewType string aliases). This is a common pattern when working with typed identifiers like MongoDB ObjectIds or UUIDs.
Note: I am not calling the function directly; the bug is caused when triggering a workflow, which, under the hood will use value_to_type
to convert types.
Describe the bug
The value_to_type
function has two bugs when handling dictionary keys with custom string types:
-
String subclasses are converted to character lists: When the key type is a string subclass (like
class CustomStr(str)
), the converter treats the string as an iterable and converts it to a list of characters, making it unhashable and unusable as a dictionary key. -
NewType fails isinstance check: When the key type is a NewType alias (like
MyNewTypeStr = NewType("MyNewTypeStr", str)
), the converter fails with "isinstance() arg 2 must be a type" because NewType creates a callable, not a proper type for isinstance checks.
Minimal Reproduction
import traceback
from dataclasses import dataclass
from typing import NewType
from temporalio.converter import value_to_type
class CustomStr(str):
pass
MyNewTypeStr = NewType("MyNewTypeStr", str)
@dataclass
class SimpleMessage:
data: dict[CustomStr, str]
@dataclass
class NewTypeMessage:
data: dict[MyNewTypeStr, str]
def test_key_conversion():
print("=" * 50)
print("Testing direct key conversion with CustomStr...")
print("=" * 50)
key = "key1"
converted_key = value_to_type(CustomStr, key)
print(f"Original key: {repr(key)} (type: {type(key)})")
print(f"Converted key: {repr(converted_key)} (type: {type(converted_key)})")
def test_newtype_key_conversion():
print("=" * 50)
print("Testing direct key conversion with NewType...")
print("=" * 50)
key = "key1"
converted_key = value_to_type(MyNewTypeStr, key)
print(f"Original key: {repr(key)} (type: {type(key)})")
print(f"Converted key: {repr(converted_key)} (type: {type(converted_key)})")
def test_full_conversion():
print("=" * 50)
print("Testing full message conversion with CustomStr...")
print("=" * 50)
payload = {"data": {"key1": "value1"}}
try:
result = value_to_type(SimpleMessage, payload)
print(f"Success: {result}")
except Exception as e:
print(f"Full conversion failed: {e}")
traceback.print_exc()
def test_newtype_full_conversion():
print("=" * 50)
print("Testing full message conversion with NewType...")
print("=" * 50)
payload = {"data": {"key1": "value1"}}
try:
result = value_to_type(NewTypeMessage, payload)
print(f"Success: {result}")
except Exception as e:
print(f"Full conversion failed: {e}")
traceback.print_exc()
if __name__ == "__main__":
test_key_conversion()
test_newtype_key_conversion()
test_full_conversion()
test_newtype_full_conversion()