feat: add TypedDict support to schema_type conversion (#91)

Adds `TypedDict` support to the to_schema method's `schema_type` parameter

Author: cofin

sqlspec/driver/mixins/_result_tools.py

@@ -7,7 +7,7 @@
 from enum import Enum
 from functools import partial
 from pathlib import Path, PurePath
-from typing import Any, Callable, Final, Optional, overload
+from typing import Any, Callable, Final, Optional, TypeVar, Union, cast, overload
 from uuid import UUID
 
 from mypy_extensions import trait
@@ -33,13 +33,16 @@
     is_dict,
     is_msgspec_struct,
     is_pydantic_model,
+    is_typed_dict,
 )
 
 __all__ = ("_DEFAULT_TYPE_DECODERS", "_default_msgspec_deserializer")
 
 
 logger = logging.getLogger(__name__)
 
+# TypeVar for TypedDict support - not bound since TypedDict classes aren't dict subclasses
+TypedDictT = TypeVar("TypedDictT")
 
 _DATETIME_TYPES: Final[set[type]] = {datetime.datetime, datetime.date, datetime.time}
 
@@ -165,10 +168,20 @@ def to_schema(data: "list[ModelT]", *, schema_type: None = None) -> "list[ModelT
     def to_schema(data: "ModelT") -> "ModelT": ...
     @overload
     @staticmethod
-    def to_schema(data: Any, *, schema_type: None = None) -> Any: ...
+    def to_schema(data: "dict[str, Any]", *, schema_type: "type[TypedDictT]") -> "TypedDictT": ...
+    @overload
+    @staticmethod
+    def to_schema(data: "list[dict[str, Any]]", *, schema_type: "type[TypedDictT]") -> "list[TypedDictT]": ...
+    @overload
+    @staticmethod
+    def to_schema(data: Any, *, schema_type: "type[TypedDictT]") -> Any: ...
 
+    @overload
     @staticmethod
-    def to_schema(data: Any, *, schema_type: "Optional[type[ModelDTOT]]" = None) -> Any:
+    def to_schema(data: Any, *, schema_type: None = None) -> Any: ...
+
+    @staticmethod  # type: ignore[misc,unused-ignore]
+    def to_schema(data: Any, *, schema_type: "Optional[type[Union[ModelDTOT, TypedDictT]]]" = None) -> Any:  # type: ignore[misc,unused-ignore]
         """Convert data to a specified schema type.
 
         Args:
@@ -183,6 +196,10 @@ def to_schema(data: Any, *, schema_type: "Optional[type[ModelDTOT]]" = None) ->
         """
         if schema_type is None:
             return data
+        if is_typed_dict(schema_type):
+            if isinstance(data, list):
+                return cast("list[ModelDTOT]", [item for item in data if is_dict(item)])
+            return cast("ModelDTOT", data)
         if is_dataclass(schema_type):
             if isinstance(data, list):
                 result: list[Any] = []
@@ -273,5 +290,5 @@ def _convert_numpy_arrays_in_data(obj: Any) -> Any:
             if isinstance(data, dict):
                 return schema_type(**data)
             return data
-        msg = "`schema_type` should be a valid Dataclass, Pydantic model, Msgspec struct, or Attrs class"
+        msg = "`schema_type` should be a valid Dataclass, Pydantic model, Msgspec struct, Attrs class, or TypedDict"
         raise SQLSpecError(msg)

sqlspec/utils/type_guards.py

@@ -9,6 +9,8 @@
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, Optional, Union, cast
 
+from typing_extensions import is_typeddict
+
 from sqlspec.typing import (
     ATTRS_INSTALLED,
     LITESTAR_INSTALLED,
@@ -117,6 +119,7 @@
     "is_select_builder",
     "is_statement_filter",
     "is_string_literal",
+    "is_typed_dict",
     "is_typed_parameter",
     "schema_dump",
     "supports_limit",
@@ -126,6 +129,18 @@
 )
 
 
+def is_typed_dict(obj: Any) -> "TypeGuard[type]":
+    """Check if an object is a TypedDict class.
+
+    Args:
+        obj: The object to check
+
+    Returns:
+        True if the object is a TypedDict class, False otherwise
+    """
+    return is_typeddict(obj)
+
+
 def is_statement_filter(obj: Any) -> "TypeGuard[StatementFilter]":
     """Check if an object implements the StatementFilter protocol.
 

tests/unit/test_driver/test_result_tools.py

@@ -10,6 +10,7 @@
 
 import msgspec
 import pytest
+from typing_extensions import TypedDict
 
 from sqlspec.driver.mixins._result_tools import (
     _DEFAULT_TYPE_DECODERS,
@@ -39,6 +40,14 @@ class SampleMsgspecStructWithIntList(msgspec.Struct):
     values: "list[int]"
 
 
+class SampleTypedDict(TypedDict):
+    """Sample TypedDict for testing."""
+
+    name: str
+    age: int
+    optional_field: "Optional[str]"
+
+
 # Test _is_list_type_target function
 def test_is_list_type_target_with_list_types() -> None:
     """Test detection of list type targets."""
@@ -290,6 +299,57 @@ def test_to_schema_mixin_without_schema_type() -> None:
     assert result == test_data
 
 
+def test_to_schema_mixin_with_typeddict_single_record() -> None:
+    """Test ToSchemaMixin.to_schema with TypedDict for single record."""
+    test_data = {"name": "test_user", "age": 30, "optional_field": "value"}
+
+    result = ToSchemaMixin.to_schema(test_data, schema_type=SampleTypedDict)
+
+    assert result == test_data
+    assert isinstance(result, dict)
+
+
+def test_to_schema_mixin_with_typeddict_multiple_records() -> None:
+    """Test ToSchemaMixin.to_schema with TypedDict for multiple records."""
+    test_data = [
+        {"name": "user1", "age": 25, "optional_field": "value1"},
+        {"name": "user2", "age": 30, "optional_field": "value2"},
+    ]
+
+    result = ToSchemaMixin.to_schema(test_data, schema_type=SampleTypedDict)
+
+    assert isinstance(result, list)
+    assert len(result) == 2
+    for item in result:
+        assert isinstance(item, dict)
+    assert result == test_data
+
+
+def test_to_schema_mixin_with_typeddict_mixed_data() -> None:
+    """Test ToSchemaMixin.to_schema with TypedDict filters non-dict items."""
+    test_data = [
+        {"name": "user1", "age": 25, "optional_field": "value1"},
+        "not_a_dict",  # This should be filtered out
+        {"name": "user2", "age": 30, "optional_field": "value2"},
+    ]
+
+    result = ToSchemaMixin.to_schema(test_data, schema_type=SampleTypedDict)
+
+    assert isinstance(result, list)
+    assert len(result) == 2  # Only dict items should be included
+    for item in result:
+        assert isinstance(item, dict)
+
+
+def test_to_schema_mixin_with_typeddict_non_dict_data() -> None:
+    """Test ToSchemaMixin.to_schema with TypedDict returns non-dict data unchanged."""
+    test_data = "not_a_dict"
+
+    result = ToSchemaMixin.to_schema(test_data, schema_type=SampleTypedDict)
+
+    assert result == test_data
+
+
 @pytest.mark.skipif(not NUMPY_INSTALLED, reason="numpy not installed")
 def test_numpy_array_conversion_edge_cases() -> None:
     """Test edge cases for numpy array conversion."""

tests/unit/test_utils/test_type_guards.py

@@ -10,6 +10,7 @@
 import msgspec
 import pytest
 from sqlglot import exp
+from typing_extensions import TypedDict
 
 from sqlspec.utils.type_guards import (
     dataclass_to_dict,
@@ -57,6 +58,7 @@
     is_schema_with_field,
     is_schema_without_field,
     is_string_literal,
+    is_typed_dict,
     schema_dump,
 )
 
@@ -74,6 +76,14 @@ class SampleDataclass:
     optional_field: "Optional[str]" = None
 
 
+class SampleTypedDict(TypedDict):
+    """Sample TypedDict for testing."""
+
+    name: str
+    age: int
+    optional_field: "Optional[str]"
+
+
 class MockSQLGlotExpression:
     """Mock SQLGlot expression for testing type guard functions.
 
@@ -932,6 +942,31 @@ def test_get_msgspec_rename_config_with_pascal_rename() -> None:
     assert result == "pascal"
 
 
+def test_is_typed_dict_with_typeddict_class() -> None:
+    """Test is_typed_dict returns True for TypedDict classes."""
+    assert is_typed_dict(SampleTypedDict) is True
+
+
+def test_is_typed_dict_with_typeddict_instance() -> None:
+    """Test is_typed_dict returns False for TypedDict instances (they are dicts)."""
+    sample_data: SampleTypedDict = {"name": "test", "age": 25, "optional_field": "value"}
+    assert is_typed_dict(sample_data) is False
+
+
+def test_is_typed_dict_with_non_typeddict() -> None:
+    """Test is_typed_dict returns False for non-TypedDict types."""
+    assert is_typed_dict(dict) is False
+    assert is_typed_dict(SampleDataclass) is False
+    assert is_typed_dict(str) is False
+    assert is_typed_dict(42) is False
+    assert is_typed_dict({}) is False
+
+
+def test_is_typed_dict_with_regular_dict() -> None:
+    """Test is_typed_dict returns False for regular dict instances."""
+    assert is_typed_dict({"key": "value"}) is False
+
+
 def test_get_msgspec_rename_config_without_rename() -> None:
     """Test get_msgspec_rename_config returns None when no rename config."""
     schema_type = MockMsgspecStructWithoutRename