fix(dataset): bound CSV export memory usage

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

src/crawlee/_utils/file.py

@@ -203,12 +203,21 @@ async def export_csv_to_stream(
     if 'lineterminator' not in kwargs:
         kwargs['lineterminator'] = '\n'
 
-    items = [item async for item in iterator if item]
-    if not items:
-        return
-
-    fieldnames = list(dict.fromkeys(key for item in items for key in item))
-    writer = csv.DictWriter(dst, fieldnames=fieldnames, **kwargs)
-    writer.writeheader()
-    for item in items:
-        writer.writerow(item)
+    fieldnames = dict[str, None]()
+    with tempfile.TemporaryFile(mode='w+', encoding='utf-8') as items:
+        async for item in iterator:
+            if not item:
+                continue
+
+            fieldnames.update(dict.fromkeys(item))
+            json.dump(item, items)
+            items.write('\n')
+
+        if not fieldnames:
+            return
+
+        writer = csv.DictWriter(dst, fieldnames=fieldnames, **kwargs)
+        writer.writeheader()
+        items.seek(0)
+        for item in items:
+            writer.writerow(json.loads(item))

tests/unit/_utils/test_file.py

@@ -63,6 +63,13 @@ async def test_export_csv_to_stream_skips_empty_items() -> None:
     assert dst.getvalue() == 'id,name\n1,Item 1\n2,Item 2\n'
 
 
+async def test_export_csv_to_stream_handles_empty_iterator() -> None:
+    dst = StringIO()
+    await export_csv_to_stream(_async_iter([]), dst)
+
+    assert dst.getvalue() == ''
+
+
 # Tests for validate_subdirectory (storage name/alias directory validation).