fixed

Author: sunxiaojian

pyiceberg/types.py

@@ -324,7 +324,7 @@ def __init__(
         self,
         field_id: Optional[int] = None,
         name: Optional[str] = None,
-        field_type: Optional[IcebergType] = None,
+        field_type: Optional[IcebergType | str] = None,
         required: bool = False,
         doc: Optional[str] = None,
         initial_default: Optional[Any] = None,
@@ -340,6 +340,12 @@ def __init__(
         data["doc"] = doc
         data["initial-default"] = data["initial-default"] if "initial-default" in data else initial_default
         data["write-default"] = data["write-default"] if "write-default" in data else write_default
+        if isinstance(data["type"], str):
+            try:
+                data["type"] = IcebergType.handle_primitive_type(data["type"], None)
+            except ValueError as e:
+                raise ValueError(f"Unsupported field type: {data['type']}.") from e
+
         super().__init__(**data)
 
     def __str__(self) -> str:

tests/test_types.py

@@ -231,6 +231,22 @@ def test_nested_field() -> None:
     assert "validation errors for NestedField" in str(exc_info.value)
 
 
+def test_nested_field_type_as_str_unsupported() -> None:
+    with pytest.raises(ValueError) as exc_info:
+        _ = (NestedField(1, "field", "list", required=True),)
+    assert "Unsupported field type: list" in str(exc_info.value)
+
+
+def test_nested_field_type_as_str_struct() -> None:
+    field_var = NestedField(
+        1,
+        "field",
+        "string",
+        required=True,
+    )
+    assert isinstance(field_var.field_type, StringType)
+
+
 @pytest.mark.parametrize("input_index,input_type", non_parameterized_types)
 @pytest.mark.parametrize("check_index,check_type", non_parameterized_types)
 def test_non_parameterized_type_equality(