Merge pull request #10 from ketgo/issue-9

Fix for parsing marshmallow nested field with many=True argument.

Author: ketgo

.github/workflows/ci.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.6", "3.7", "3.8"]
+        python-version: ["3.7", "3.8"]
 
     steps:
     - uses: actions/checkout@v3

.gitignore

@@ -73,9 +73,12 @@ fabric.properties
 
 .idea/*
 
+# Python files and dirs to ignore
 venv/*
-
 *.egg-info
+*.pyc
+__pycache__/
+dist/
 
 .coverage
 

marshmallow_pyspark/converters.py

@@ -156,4 +156,4 @@ def convert(self, ma_field: ma_fields.Nested) -> DataType:
                     nullable=True
                 )
             )
-        return StructType(_fields)
+        return ArrayType(StructType(_fields)) if ma_field.many else StructType(_fields)

marshmallow_pyspark/version.py

@@ -2,7 +2,7 @@
 Version for marshmallow_pyspark package
 """
 
-__version__ = '0.2.3'  # pragma: no cover
+__version__ = '0.2.4'  # pragma: no cover
 
 
 def version_info():  # pragma: no cover

tests/test_schema.py

@@ -29,7 +29,9 @@ def test_create():
     (fields.Integer(), IntegerType()),
     (fields.Number(), DoubleType()),
     (fields.List(fields.String()), ArrayType(StringType())),
-    (fields.Nested(Schema.from_dict({"name": fields.String()})), StructType([StructField("name", StringType())]))
+    (fields.Nested(Schema.from_dict({"name": fields.String()})), StructType([StructField("name", StringType())])),
+    (fields.Nested(Schema.from_dict({"name": fields.String()}), many=True),
+     ArrayType(StructType([StructField("name", StringType())])))
 ])
 def test_spark_schema(ma_field, spark_field):
     class TestSchema(Schema):
@@ -110,6 +112,26 @@ class TestSchema(Schema):
             [
                 {"name": "invalid_1", "book": {"author": "Sam", "title": "Sam's Book", "cost": "32a"}},
             ]
+    ),
+    (
+            Schema.from_dict({
+                "name": fields.String(required=True),
+                "book": fields.Nested(
+                    Schema.from_dict({
+                        "author": fields.String(required=True),
+                        "title": fields.String(required=True),
+                        "cost": fields.Number(required=True)
+                    }),
+                    many=True
+                )
+            }),
+            [
+                {"name": "valid_1", "book": [{"author": "Sam", "title": "Sam's Book", "cost": "32.5"}]},
+            ],
+            [
+                {"name": "valid_1", "book": [{"author": "Sam", "title": "Sam's Book", "cost": 32.5}]},
+            ],
+            []
     )
 ])
 def test_validate_df(spark_session, schema, input_data, valid_rows, invalid_rows):