refactor(incremental): introduce StreamRecord type

Replicates graphql/graphql-js@eceeb4c

Author: Cito

src/graphql/execution/__init__.py

@@ -23,6 +23,7 @@
     InitialIncrementalExecutionResult,
     PendingResult,
     SubsequentIncrementalExecutionResult,
+    StreamRecord,
 )
 from .middleware import MiddlewareManager
 from .values import get_argument_values, get_directive_values, get_variable_values
@@ -60,6 +61,7 @@
     "Middleware",
     "MiddlewareManager",
     "PendingResult",
+    "StreamRecord",
     "SubsequentIncrementalExecutionResult",
     "create_source_event_stream",
     "default_field_resolver",

src/graphql/execution/execute.py

@@ -104,7 +104,7 @@
     ReconcilableStreamItemsResult,
     StreamItemsRecord,
     StreamItemsResult,
-    SubsequentResultRecord,
+    StreamRecord,
     TerminatingStreamItemsResult,
     is_reconcilable_stream_items_result,
 )
@@ -957,10 +957,10 @@ async def complete_async_iterator_value(
                     early_return = async_iterator.aclose()  # type: ignore
                 except AttributeError:
                     early_return = None
-                stream_record: SubsequentResultRecord | CancellableStreamRecord
+                stream_record: StreamRecord
 
                 if early_return is None:
-                    stream_record = SubsequentResultRecord(path, stream_usage.label)
+                    stream_record = StreamRecord(path, stream_usage.label)
                 else:
                     stream_record = CancellableStreamRecord(
                         early_return, path, stream_usage.label
@@ -1117,7 +1117,7 @@ def complete_iterable_value(
             except StopIteration:
                 break
             if stream_usage and index >= stream_usage.initial_count:
-                stream_record = SubsequentResultRecord(path, stream_usage.label)
+                stream_record = StreamRecord(path, stream_usage.label)
 
                 first_stream_items = self.first_sync_stream_items(
                     stream_record,
@@ -1726,7 +1726,7 @@ async def await_result() -> DeferredGroupedFieldSetResult:
 
     def first_sync_stream_items(
         self,
-        stream_record: SubsequentResultRecord,
+        stream_record: StreamRecord,
         initial_item: AwaitableOrValue[Any],
         initial_index: int,
         iterator: Iterable[Any],
@@ -1798,7 +1798,7 @@ async def await_result() -> StreamItemsResult:
 
     def first_async_stream_items(
         self,
-        stream_record: SubsequentResultRecord,
+        stream_record: StreamRecord,
         path: Path,
         initial_index: int,
         async_iterator: AsyncIterator[Any],
@@ -1822,7 +1822,7 @@ def first_async_stream_items(
 
     async def get_next_async_stream_items_result(
         self,
-        stream_record: SubsequentResultRecord,
+        stream_record: StreamRecord,
         path: Path,
         index: int,
         async_iterator: AsyncIterator[Any],
@@ -1868,7 +1868,7 @@ async def get_next_async_stream_items_result(
 
     def complete_stream_items(
         self,
-        stream_record: SubsequentResultRecord,
+        stream_record: StreamRecord,
         item_path: Path,
         item: Any,
         incremental_context: IncrementalContext,
@@ -2371,7 +2371,7 @@ def prepend_next_resolved_stream_items(
 
 def build_stream_items_result(
     errors: list[GraphQLError] | None,
-    stream_record: SubsequentResultRecord,
+    stream_record: StreamRecord,
     result: GraphQLWrappedResult[Any],
 ) -> StreamItemsResult:
     """Build a stream items result."""

src/graphql/execution/incremental_graph.py

@@ -16,7 +16,7 @@
 )
 
 from graphql.execution.types import (
-    SubsequentResultRecord,
+    StreamRecord,
     is_deferred_grouped_field_set_record,
 )
 
@@ -31,6 +31,7 @@
         ReconcilableDeferredGroupedFieldSetResult,
         StreamItemsRecord,
         StreamItemsResult,
+        SubsequentResultRecord,
     )
 
     try:
@@ -67,7 +68,7 @@ def __init__(self, deferred_fragment_record: DeferredFragmentRecord) -> None:
         self.children = []
 
 
-SubsequentResultNode = Union[DeferredFragmentNode, SubsequentResultRecord]
+SubsequentResultNode = Union[DeferredFragmentNode, StreamRecord]
 
 
 def is_deferred_fragment_node(
@@ -79,9 +80,9 @@ def is_deferred_fragment_node(
 
 def is_stream_node(
     node: SubsequentResultNode | None,
-) -> TypeGuard[SubsequentResultRecord]:
+) -> TypeGuard[StreamRecord]:
     """Check whether the given result node is a stream node."""
-    return isinstance(node, SubsequentResultRecord)
+    return isinstance(node, StreamRecord)
 
 
 class IncrementalGraph:
@@ -251,7 +252,7 @@ def remove_deferred_fragment(
         for child in deferred_fragment_node.children:  # pragma: no cover
             self.remove_deferred_fragment(child.deferred_fragment_record)
 
-    def remove_stream(self, stream_record: SubsequentResultRecord) -> None:
+    def remove_stream(self, stream_record: StreamRecord) -> None:
         """Remove a stream record as no longer pending."""
         self._remove_pending(stream_record)
 

src/graphql/execution/types.py

@@ -799,18 +799,25 @@ def __init__(
         self.deferred_fragment_records = deferred_fragment_records
 
 
-class SubsequentResultRecord:
-    """Subsequent result record"""
+class DeferredFragmentRecord:
+    """Deferred fragment record"""
 
     path: Path | None
     label: str | None
     id: str | None
+    parent: DeferredFragmentRecord | None
 
-    __slots__ = "id", "label", "path"
+    __slots__ = "id", "label", "parent", "path"
 
-    def __init__(self, path: Path | None, label: str | None = None) -> None:
+    def __init__(
+        self,
+        path: Path | None = None,
+        label: str | None = None,
+        parent: DeferredFragmentRecord | None = None,
+    ) -> None:
         self.path = path
         self.label = label
+        self.parent = parent
         self.id = None
 
     def __repr__(self) -> str:
@@ -823,38 +830,44 @@ def __repr__(self) -> str:
         return f"{name}({', '.join(args)})"
 
 
-class DeferredFragmentRecord(SubsequentResultRecord):
-    """Deferred fragment record
+class StreamRecord:
+    """Stream record"""
 
-    For internal use only.
-    """
-
-    parent: DeferredFragmentRecord | None
+    path: Path
+    label: str | None
+    id: str | None
 
-    __slots__ = ("parent",)
+    __slots__ = "id", "label", "path"
 
     def __init__(
         self,
-        path: Path | None = None,
+        path: Path,
         label: str | None = None,
-        parent: DeferredFragmentRecord | None = None,
     ) -> None:
-        super().__init__(path, label)
-        self.parent = parent
+        self.path = path
+        self.label = label
+        self.id = None
+
+    def __repr__(self) -> str:
+        name = self.__class__.__name__
+        args: list[str] = [f"path={self.path.as_list()!r}"]
+        if self.label:
+            args.append(f"label={self.label!r}")
+        return f"{name}({', '.join(args)})"
+
 
+SubsequentResultRecord = Union[DeferredFragmentRecord, StreamRecord]
 
-class CancellableStreamRecord(SubsequentResultRecord):
+
+class CancellableStreamRecord(StreamRecord):
     """Cancellable stream record"""
 
     early_return: Awaitable[None]
 
     __slots__ = ("early_return",)
 
     def __init__(
-        self,
-        early_return: Awaitable[None],
-        path: Path | None = None,
-        label: str | None = None,
+        self, early_return: Awaitable[None], path: Path, label: str | None = None
     ) -> None:
         super().__init__(path, label)
         self.early_return = early_return
@@ -870,7 +883,7 @@ def is_cancellable_stream_record(
 class ReconcilableStreamItemsResult(NamedTuple):
     """Reconcilable stream items result"""
 
-    stream_record: SubsequentResultRecord
+    stream_record: StreamRecord
     result: BareStreamItemsResult
     incremental_data_records: list[IncrementalDataRecord] | None = None
     errors: None = None
@@ -886,7 +899,7 @@ def is_reconcilable_stream_items_result(
 class TerminatingStreamItemsResult(NamedTuple):
     """Terminating stream items result"""
 
-    stream_record: SubsequentResultRecord
+    stream_record: StreamRecord
     result: None = None
     incremental_data_record: None = None
     errors: None = None
@@ -895,7 +908,7 @@ class TerminatingStreamItemsResult(NamedTuple):
 class NonReconcilableStreamItemsResult(NamedTuple):
     """Non-reconcilable stream items result"""
 
-    stream_record: SubsequentResultRecord
+    stream_record: StreamRecord
     errors: list[GraphQLError]
     result: None = None
 
@@ -912,12 +925,12 @@ class StreamItemsRecord:
 
     __slots__ = "result", "stream_record"
 
-    stream_record: SubsequentResultRecord
+    stream_record: StreamRecord
     result: AwaitableOrValue[StreamItemsResult]
 
     def __init__(
         self,
-        stream_record: SubsequentResultRecord,
+        stream_record: StreamRecord,
         result: AwaitableOrValue[StreamItemsResult],
     ) -> None:
         self.stream_record = stream_record

tests/execution/test_stream.py

@@ -10,9 +10,11 @@
     ExecutionResult,
     ExperimentalIncrementalExecutionResults,
     IncrementalStreamResult,
+    StreamRecord,
     experimental_execute_incrementally,
 )
 from graphql.language import DocumentNode, parse
+from graphql.pyutils import Path
 from graphql.type import (
     GraphQLField,
     GraphQLID,
@@ -244,6 +246,14 @@ def can_hash_incremental_stream_result():
             IncrementalStreamResult(**modified_args(args, extensions={"baz": 1}))
         )
 
+    def can_print_stream_record():
+        """Can print a StreamRecord"""
+        path = Path(None, "bar", "Bar")
+        record = StreamRecord(path)
+        assert str(record) == "StreamRecord(path=['bar'])"
+        record = StreamRecord(path, "foo")
+        assert str(record) == "StreamRecord(path=['bar'], label='foo')"
+
     async def can_stream_a_list_field():
         """Can stream a list field"""
         document = parse("{ scalarList @stream(initialCount: 1) }")