incremental: avoid double loop with stream from sync iterables

Replicates graphql/graphql-js@d245e65

Author: Cito

pyproject.toml

@@ -275,6 +275,7 @@ disallow_untyped_defs = false
 [tool.pyright]
 # silence pyright since we're using mypy already
 reportArgumentType = false
+reportAttributeAccessIssue = false
 reportIncompatibleVariableOverride = false
 reportInvalidTypeForm = false
 reportMissingModuleSource = false

src/graphql/execution/execute.py

@@ -1717,11 +1717,13 @@ def first_sync_stream_items(
         item_type: GraphQLOutputType,
     ) -> StreamItemsRecord:
         """Get the first sync stream items."""
-        is_awaitable = self.is_awaitable
-        path = stream_record.path
-        initial_path = Path(path, initial_index, None)
 
         async def await_result() -> StreamItemsResult:
+            is_awaitable = self.is_awaitable
+            prepend_next_stream_items = self.prepend_next_stream_items
+            path = stream_record.path
+            initial_path = Path(path, initial_index, None)
+
             result = self.complete_stream_items(
                 stream_record,
                 initial_path,
@@ -1731,8 +1733,8 @@ async def await_result() -> StreamItemsResult:
                 info,
                 item_type,
             )
-            results = [result]
-            append_result = results.append
+            first_stream_items = StreamItemsRecord(stream_record, result)
+            current_stream_items = first_stream_items
             current_index = initial_index
             errored_synchronously = False
             for item in iterator:
@@ -1752,33 +1754,27 @@ async def await_result() -> StreamItemsResult:
                     info,
                     item_type,
                 )
-                append_result(result)
 
-            current_index = len(results) - 1
+                next_stream_items = StreamItemsRecord(stream_record, result)
+                current_stream_items.result = prepend_next_stream_items(
+                    current_stream_items.result, next_stream_items
+                )
+                current_stream_items = next_stream_items
+
             # If a non-reconcilable stream items result was encountered,
             # then the stream terminates in error. Otherwise, add a stream terminator.
-            prepend_next_stream_items = self.prepend_next_stream_items
-            current_result = (
-                results[current_index]
-                if errored_synchronously
-                else prepend_next_stream_items(
-                    results[current_index],
+            if not errored_synchronously:
+                current_stream_items.result = prepend_next_stream_items(
+                    current_stream_items.result,
                     StreamItemsRecord(
                         stream_record, TerminatingStreamItemsResult(stream_record)
                     ),
                 )
-            )
 
-            while current_index > 0:
-                current_index -= 1
-                current_result = prepend_next_stream_items(
-                    results[current_index],
-                    StreamItemsRecord(stream_record, current_result),
-                )
-
-            if is_awaitable(current_result):
-                return await current_result  # type: ignore
-            return current_result  # type: ignore
+            result = first_stream_items.result
+            if is_awaitable(result):
+                return await result  # type: ignore
+            return result  # type: ignore
 
         return StreamItemsRecord(stream_record, await_result())
 

src/graphql/execution/incremental_publisher.py

@@ -938,12 +938,22 @@ class NonReconcilableStreamItemsResult(NamedTuple):
 ]
 
 
-class StreamItemsRecord(NamedTuple):
+class StreamItemsRecord:
     """Stream items record"""
 
+    __slots__ = "result", "stream_record"
+
     stream_record: SubsequentResultRecord
     result: AwaitableOrValue[StreamItemsResult]
 
+    def __init__(
+        self,
+        stream_record: SubsequentResultRecord,
+        result: AwaitableOrValue[StreamItemsResult],
+    ) -> None:
+        self.stream_record = stream_record
+        self.result = result
+
 
 IncrementalDataRecord = Union[DeferredGroupedFieldSetRecord, StreamItemsRecord]