Fix/3159 pydantic model incorrect serialization

dlt/extract/validation.py

@@ -40,18 +40,29 @@ def __call__(self, item: TDataItems, meta: Any = None) -> TDataItems:
         from dlt.common.libs.pydantic import validate_and_filter_item, validate_and_filter_items
 
         if isinstance(item, list):
-            return [
-                model.dict(by_alias=True)
-                for model in validate_and_filter_items(
-                    self.table_name, self.list_model, item, self.column_mode, self.data_mode
-                )
-            ]
-        item = validate_and_filter_item(
+            input_is_model = bool(item) and isinstance(item[0], PydanticBaseModel)
+            validated_list = validate_and_filter_items(
+                self.table_name, self.list_model, item, self.column_mode, self.data_mode
+            )
+            if input_is_model:
+                input_fields = set(item[0].__class__.model_fields.keys())
+                validated_fields = set(validated_list[0].__class__.model_fields.keys())
+                if input_fields.issubset(validated_fields):
+                    return validated_list
+            return [m.dict(by_alias=True) for m in validated_list]
+
+        input_is_model = isinstance(item, PydanticBaseModel)
+        validated = validate_and_filter_item(
             self.table_name, self.model, item, self.column_mode, self.data_mode
         )
-        if item is not None:
-            item = item.dict(by_alias=True)
-        return item
+        if validated is None:
+            return None
+        if input_is_model:
+            input_fields = set(item.__class__.model_fields.keys())
+            validated_fields = set(validated.__class__.model_fields.keys())
+            if input_fields.issubset(validated_fields):
+                return validated
+        return validated.dict(by_alias=True)
 
     def __str__(self, *args: Any, **kwargs: Any) -> str:
         return f"PydanticValidator(model={self.model.__qualname__})"
@@ -93,4 +104,4 @@ def create_item_validator(
             ),
             schema_contract or expanded_schema_contract,
         )
-    return None, schema_contract
+    return None, schema_contract

tests/pipeline/test_pipeline_extra.py

@@ -855,3 +855,53 @@ def inconsistent_data(dtype: str):
         # generates variant column on non-nullable column. original "foo" will receive null
         pipeline.run(inconsistent_data("text"))
     assert pip_ex.value.step == "normalize"
+
+
+@pytest.mark.parametrize(
+    "as_model, as_list",
+    [
+        (False, False),
+        (True, False),
+        (False, True),
+        (True, True),
+    ],
+)
+def test_pydantic_validator_preserves_model_instances(as_model, as_list):
+    class Result(BaseModel):
+        number: int
+
+    @dlt.resource(columns=Result)
+    def data():
+        if as_model:
+            item = Result(number=1)
+        else:
+            item = {"number": 1}# type: ignore[assignment]
+        if as_list:
+            yield [item, item, item]
+        else:
+            yield item
+
+    seen = []
+
+    @dlt.transformer(data_from=data)
+    def check(x):
+        seen.append(x)
+        yield x
+
+    pipeline = dlt.pipeline(destination="duckdb", dev_mode=True)
+    pipeline.run(check)
+
+    assert len(seen) == 1
+    v = seen[0]
+
+    if as_list:
+        assert isinstance(v, list)
+        if as_model:
+            assert all(isinstance(el, BaseModel) for el in v)
+        else:
+            assert all(isinstance(el, dict) for el in v)
+    else:
+        if as_model:
+            assert isinstance(v, BaseModel)
+        else:
+            assert isinstance(v, dict)