Repository: frictionlessdata/frictionless-py

Issue:
Add streaming support to the extract functions
# Overview

we need an option to return a stream instead of loading into memory.

Discussion:

Patch:
diff --git a/frictionless/extract/main.py b/frictionless/extract/main.py
index ac888f76..20d0422b 100644
--- a/frictionless/extract/main.py
+++ b/frictionless/extract/main.py
@@ -6,9 +6,8 @@ from ..package import Package
 from .. import helpers
 
 
-# NOTE: ability to extract stream?
-def extract(source, *, source_type=None, process=None, **options):
-    """Extract resource rows into memory
+def extract(source, *, source_type=None, process=None, stream=False, **options):
+    """Extract resource rows
 
     API      | Usage
     -------- | --------
@@ -18,6 +17,7 @@ def extract(source, *, source_type=None, process=None, **options):
         source (dict|str): data source
         source_type (str): source type - package, resource or table
         process? (func): a row processor function
+        stream? (bool): return a row stream(s) instead of loading into memory
         **options (dict): options for the underlaying function
 
     Returns:
@@ -46,4 +46,4 @@ def extract(source, *, source_type=None, process=None, **options):
 
     # Extract source
     extract = getattr(module, "extract_%s" % source_type)
-    return extract(source, process=process, **options)
+    return extract(source, process=process, stream=stream, **options)
diff --git a/frictionless/extract/package.py b/frictionless/extract/package.py
index 1ebd2540..83e378e1 100644
--- a/frictionless/extract/package.py
+++ b/frictionless/extract/package.py
@@ -2,8 +2,8 @@ from collections import OrderedDict
 from ..package import Package
 
 
-def extract_package(source, *, process=None):
-    """Extract package rows into memory
+def extract_package(source, *, process=None, stream=False):
+    """Extract package rows
 
     API      | Usage
     -------- | --------
@@ -12,9 +12,10 @@ def extract_package(source, *, process=None):
     Parameters:
         source (dict|str): data resource descriptor
         process? (func): a row processor function
+        stream? (bool): return a row streams instead of loading into memory
 
     Returns:
-        {path: Row[]}: a dictionary of arrays of rows
+        {path: Row[]}: a dictionary of arrays/streams of rows
 
     """
 
@@ -25,10 +26,7 @@ def extract_package(source, *, process=None):
     result = OrderedDict()
     for number, resource in enumerate(package.resources, start=1):
         key = resource.fullpath if not resource.inline else f"memory{number}"
-        if process:
-            result[key] = []
-            for row in resource.read_row_stream():
-                result[key].append(process(row))
-            continue
-        result[key] = resource.read_rows()
+        data = resource.read_row_stream()
+        data = (process(row) for row in data) if process else data
+        result[key] = data if stream else list(data)
     return result
diff --git a/frictionless/extract/resource.py b/frictionless/extract/resource.py
index 0ca12d6a..e2a547d1 100644
--- a/frictionless/extract/resource.py
+++ b/frictionless/extract/resource.py
@@ -1,8 +1,8 @@
 from ..resource import Resource
 
 
-def extract_resource(source, *, process=None):
-    """Extract resource rows into memory
+def extract_resource(source, *, process=None, stream=False):
+    """Extract resource rows
 
     API      | Usage
     -------- | --------
@@ -13,7 +13,7 @@ def extract_resource(source, *, process=None):
         process? (func): a row processor function
 
     Returns:
-        Row[]: an array of rows
+        Row[]: an array/stream of rows
 
     """
 
@@ -21,9 +21,6 @@ def extract_resource(source, *, process=None):
     resource = Resource(source)
 
     # Extract resource
-    if process:
-        result = []
-        for row in resource.read_row_stream():
-            result.append(process(row))
-        return result
-    return resource.read_rows()
+    data = resource.read_row_stream()
+    data = (process(row) for row in data) if process else data
+    return data if stream else list(data)
diff --git a/frictionless/extract/table.py b/frictionless/extract/table.py
index b0c13fad..d258f81e 100644
--- a/frictionless/extract/table.py
+++ b/frictionless/extract/table.py
@@ -28,8 +28,9 @@ def extract_table(
     lookup=None,
     # Extraction
     process=None,
+    stream=False,
 ):
-    """Extract table rows into memory
+    """Extract table rows
 
     API      | Usage
     -------- | --------
@@ -112,9 +113,10 @@ def extract_table(
             For more information, please check "Extracting  Data" guide.
 
         process? (func): a row processor function
+        stream? (bool): return a row streams instead of loading into memory
 
     Returns:
-        Row[]: an array for rows
+        Row[]: an array/stream of rows
 
     """
 
@@ -144,10 +146,15 @@ def extract_table(
     )
 
     # Extract table
+    data = read_row_stream(table)
+    data = (process(row) for row in data) if process else data
+    return data if stream else list(data)
+
+
+# Internal
+
+
+def read_row_stream(table):
     with table as table:
-        if process:
-            result = []
-            for row in table.row_stream:
-                result.append(process(row))
-            return result
-        return table.read_rows()
+        for row in table.row_stream:
+            yield row


Test patch:
diff --git a/tests/extract/test_package.py b/tests/extract/test_package.py
index 8560e9a0..0ebbc185 100644
--- a/tests/extract/test_package.py
+++ b/tests/extract/test_package.py
@@ -1,3 +1,4 @@
+import types
 from frictionless import extract
 
 
@@ -8,3 +9,34 @@ def test_extract_package():
     assert extract("data/package.json") == {
         "data/table.csv": [{"id": 1, "name": "english"}, {"id": 2, "name": "中国人"}]
     }
+
+
+def test_extract_package_process():
+    process = lambda row: row.to_list()
+    assert extract("data/package.json", process=process) == {
+        "data/table.csv": [
+            [1, "english"],
+            [2, "中国人"],
+        ],
+    }
+
+
+def test_extract_package_stream():
+    row_streams = extract("data/package.json", stream=True)
+    row_stream = row_streams["data/table.csv"]
+    assert isinstance(row_stream, types.GeneratorType)
+    assert list(row_stream) == [
+        {"id": 1, "name": "english"},
+        {"id": 2, "name": "中国人"},
+    ]
+
+
+def test_extract_package_process_and_stream():
+    process = lambda row: row.to_list()
+    data_streams = extract("data/package.json", process=process, stream=True)
+    data_stream = data_streams["data/table.csv"]
+    assert isinstance(data_stream, types.GeneratorType)
+    assert list(data_stream) == [
+        [1, "english"],
+        [2, "中国人"],
+    ]
diff --git a/tests/extract/test_resource.py b/tests/extract/test_resource.py
index 93bb8c1e..87971a52 100644
--- a/tests/extract/test_resource.py
+++ b/tests/extract/test_resource.py
@@ -1,3 +1,4 @@
+import types
 from frictionless import extract
 
 
@@ -17,3 +18,22 @@ def test_extract_resource_process():
         [1, "english"],
         [2, "中国人"],
     ]
+
+
+def test_extract_resource_stream():
+    row_stream = extract("data/resource.json", stream=True)
+    assert isinstance(row_stream, types.GeneratorType)
+    assert list(row_stream) == [
+        {"id": 1, "name": "english"},
+        {"id": 2, "name": "中国人"},
+    ]
+
+
+def test_extract_table_process_and_stream():
+    process = lambda row: row.to_list()
+    data_stream = extract("data/resource.json", process=process, stream=True)
+    assert isinstance(data_stream, types.GeneratorType)
+    assert list(data_stream) == [
+        [1, "english"],
+        [2, "中国人"],
+    ]
diff --git a/tests/extract/test_table.py b/tests/extract/test_table.py
index 51adb8c2..00c2e81f 100644
--- a/tests/extract/test_table.py
+++ b/tests/extract/test_table.py
@@ -1,3 +1,4 @@
+import types
 from frictionless import extract
 
 
@@ -9,3 +10,30 @@ def test_extract_table():
         {"id": 1, "name": "english"},
         {"id": 2, "name": "中国人"},
     ]
+
+
+def test_extract_table_process():
+    process = lambda row: row.to_list()
+    assert extract("data/table.csv", process=process) == [
+        [1, "english"],
+        [2, "中国人"],
+    ]
+
+
+def test_extract_table_stream():
+    row_stream = extract("data/table.csv", stream=True)
+    assert isinstance(row_stream, types.GeneratorType)
+    assert list(row_stream) == [
+        {"id": 1, "name": "english"},
+        {"id": 2, "name": "中国人"},
+    ]
+
+
+def test_extract_table_process_and_stream():
+    process = lambda row: row.to_list()
+    data_stream = extract("data/table.csv", process=process, stream=True)
+    assert isinstance(data_stream, types.GeneratorType)
+    assert list(data_stream) == [
+        [1, "english"],
+        [2, "中国人"],
+    ]