✨ StreamParsing - Add streaming parsing capacity

magicparse/__init__.py

@@ -1,11 +1,12 @@
 from io import BytesIO
+from typing import Any, Dict, List, Tuple, Union
 
+from .callbacks import OnInvalidRowCallback, OnValidRowCallback
 from .schema import Schema, builtins as builtins_schemas
 from .post_processors import PostProcessor, builtins as builtins_post_processors
 from .pre_processors import PreProcessor, builtins as builtins_pre_processors
 from .transform import Transform
 from .type_converters import TypeConverter, builtins as builtins_type_converters
-from typing import Any, Dict, List, Tuple, Union
 from .validators import Validator, builtins as builtins_validators
 
 
@@ -26,6 +27,20 @@ def parse(
     return schema_definition.parse(data)
 
 
+def stream_parse(
+    data: Union[bytes, BytesIO],
+    schema_options: Dict[str, Any],
+    on_valid_parsed_row: OnValidRowCallback,
+    on_invalid_parsed_row: OnInvalidRowCallback,
+) -> None:
+    schema_definition = Schema.build(schema_options)
+    return schema_definition.stream_parse(
+        data=data,
+        on_valid_parsed_row=on_valid_parsed_row,
+        on_invalid_parsed_row=on_invalid_parsed_row,
+    )
+
+
 Registrable = Union[Schema, Transform]
 
 

magicparse/callbacks.py

@@ -0,0 +1,11 @@
+from typing import Any, Dict, List, Protocol
+
+
+class OnValidRowCallback(Protocol):
+    def __call__(self, index: int, parsed_row: Dict[str, Any], raw_data: Any) -> None:
+        ...
+
+
+class OnInvalidRowCallback(Protocol):
+    def __call__(self, errors_info: List[Dict[str, Any]], raw_data: Any) -> None:
+        ...

magicparse/schema.py

@@ -1,6 +1,8 @@
 import codecs
 from abc import ABC, abstractmethod
 import csv
+
+from magicparse import OnInvalidRowCallback, OnValidRowCallback
 from .fields import Field
 from io import BytesIO
 from typing import Any, Dict, List, Tuple, Union, Iterable
@@ -75,6 +77,45 @@ def parse(self, data: Union[bytes, BytesIO]) -> Tuple[List[dict], List[dict]]:
 
         return result, errors
 
+    def stream_parse(
+        self,
+        data: Union[bytes, BytesIO],
+        on_valid_parsed_row: OnValidRowCallback,
+        on_invalid_parsed_row: OnInvalidRowCallback,
+    ) -> None:
+        if isinstance(data, bytes):
+            stream = BytesIO(data)
+        else:
+            stream = data
+
+        reader = self.get_reader(stream)
+
+        row_number = 0
+        if self.has_header:
+            next(reader)
+            row_number += 1
+
+        for row in reader:
+            errors = []
+            row_is_valid = True
+            item = {}
+            for field in self.fields:
+                try:
+                    value = field.read_value(row)
+                except Exception as exc:
+                    errors.append({"row-number": row_number, **field.error(exc)})
+                    row_is_valid = False
+                    continue
+
+                item[field.key] = value
+
+            if row_is_valid:
+                on_valid_parsed_row(index=row_number, parsed_row=item, raw_data=row)
+            else:
+                on_invalid_parsed_row(errors_info=errors, raw_data=row)
+
+            row_number += 1
+
 
 class CsvSchema(Schema):
     def __init__(self, options: Dict[str, Any]) -> None: