From 1b3cccc31e08bb76eede3c0fdbdf51865e364f26 Mon Sep 17 00:00:00 2001 From: Evan Erwee Date: Wed, 15 Jul 2026 16:24:15 -0400 Subject: [PATCH] feat: add S3Checkpoint for serverless/containerized environments Add S3-backed checkpoint that stores markers as zero-byte S3 objects instead of local files. Enables checkpointing in ECS Fargate, Lambda, and EKS environments where local disk is ephemeral. - S3CheckpointFilter: HEAD request to check if node was processed - S3CheckpointWriter: PUT zero-byte object as completion marker - S3Checkpoint: drop-in replacement for Checkpoint with same interface (add_filter, add_writer, enabled) The local Checkpoint class uses os.path.exists/os.makedirs/open which requires persistent local filesystem. S3Checkpoint uses the same pattern but backed by S3, surviving container restarts and retries. Usage: from graphrag_toolkit.lexical_graph.indexing.build import S3Checkpoint checkpoint = S3Checkpoint( checkpoint_name='enrichment-run-001', bucket_name='my-pipeline-bucket', key_prefix='checkpoints/tenant-a' ) graph_index.extract(docs, handler=extracted_docs, checkpoint=checkpoint) --- .../lexical_graph/indexing/build/__init__.py | 1 + .../indexing/build/s3_checkpoint.py | 305 ++++++++++++++++ .../unit/indexing/build/test_s3_checkpoint.py | 327 ++++++++++++++++++ 3 files changed, 633 insertions(+) create mode 100644 lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/s3_checkpoint.py create mode 100644 lexical-graph/tests/unit/indexing/build/test_s3_checkpoint.py diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/__init__.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/__init__.py index 58e14bdd3..d41ceca91 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/__init__.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/__init__.py @@ -6,5 +6,6 @@ from .graph_construction import GraphConstruction from .version_manager import VersionManager from .checkpoint import Checkpoint +from .s3_checkpoint import S3Checkpoint from .build_filters import BuildFilters, DEFAULT_BUILD_FILTER from .delete_sources import DeletePrevVersions \ No newline at end of file diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/s3_checkpoint.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/s3_checkpoint.py new file mode 100644 index 000000000..eff6c1c35 --- /dev/null +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/indexing/build/s3_checkpoint.py @@ -0,0 +1,305 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +import logging +from typing import Any, List + +import boto3 +from botocore.exceptions import ClientError + +from graphrag_toolkit.lexical_graph.tenant_id import TenantId +from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler +from graphrag_toolkit.lexical_graph.indexing.build.checkpoint import DoNotCheckpoint +from graphrag_toolkit.lexical_graph.storage.constants import INDEX_KEY + +from llama_index.core.schema import TransformComponent, BaseNode + +SAVEPOINT_ROOT_DIR = 'save_points' + +logger = logging.getLogger(__name__) + + +class S3CheckpointFilter(TransformComponent, DoNotCheckpoint): + """Filters nodes based on the absence of an S3 checkpoint marker. + + S3-backed equivalent of CheckpointFilter. Uses HEAD requests on zero-byte + S3 objects instead of os.path.exists() on local files. + + Attributes: + checkpoint_name (str): The name of the checkpoint used for filtering. + bucket_name (str): S3 bucket where checkpoint markers are stored. + key_prefix (str): S3 key prefix for checkpoint markers. + inner (TransformComponent): The wrapped TransformComponent for processing nodes. + tenant_id (TenantId): Tenant ID for multi-tenancy support. + """ + checkpoint_name: str + bucket_name: str + key_prefix: str + inner: TransformComponent + tenant_id: TenantId + + class Config: + arbitrary_types_allowed = True + + def _get_s3_client(self): + from graphrag_toolkit.lexical_graph.config import GraphRAGConfig + return GraphRAGConfig.s3 + + def _marker_key(self, node_id: str) -> str: + return f"{self.key_prefix}/{SAVEPOINT_ROOT_DIR}/{self.checkpoint_name}/{node_id}" + + def checkpoint_does_not_exist(self, node_id: str) -> bool: + """Check whether a checkpoint marker exists in S3 for the given node. + + Args: + node_id: Identifier of the node to check. + + Returns: + bool: True if no checkpoint exists (node should be processed), + False if checkpoint exists (node should be skipped). + """ + tenant_node_id = self.tenant_id.rewrite_id(node_id) + key = self._marker_key(tenant_node_id) + + try: + s3_client = self._get_s3_client() + s3_client.head_object(Bucket=self.bucket_name, Key=key) + logger.debug( + f'Ignoring node because checkpoint already exists ' + f'[node_id: {tenant_node_id}, checkpoint: {self.checkpoint_name}, ' + f'component: {type(self.inner).__name__}]' + ) + return False + except ClientError as e: + if e.response['Error']['Code'] in ('404', 'NoSuchKey'): + logger.debug( + f'Including node ' + f'[node_id: {tenant_node_id}, checkpoint: {self.checkpoint_name}, ' + f'component: {type(self.inner).__name__}]' + ) + return True + # Unexpected error — include node (safe default: re-process rather than skip) + logger.warning( + f'Checkpoint check failed for node {tenant_node_id}: {e}. ' + f'Including node (safe default).' + ) + return True + + def __call__(self, nodes: List[BaseNode], **kwargs: Any) -> List[BaseNode]: + """Filter nodes that already have S3 checkpoint markers, then process the rest. + + Args: + nodes: A list of BaseNode objects to be filtered. + **kwargs: Additional keyword arguments passed to the inner callable. + + Returns: + A list of BaseNode objects filtered and processed by the inner callable. + """ + discarded_count = 0 + filtered_nodes = [] + + for node in nodes: + if self.checkpoint_does_not_exist(node.id_): + filtered_nodes.append(node) + else: + discarded_count += 1 + + if discarded_count > 0: + logger.info( + f'[{type(self.inner).__name__}] Discarded {discarded_count} out of ' + f'{discarded_count + len(filtered_nodes)} nodes because they have ' + f'already been checkpointed (S3)' + ) + + return self.inner.__call__(filtered_nodes, **kwargs) + + +class S3CheckpointWriter(NodeHandler): + """Writes S3 checkpoint markers for processed nodes. + + S3-backed equivalent of CheckpointWriter. Uses PUT zero-byte objects + instead of touching local files. + + Attributes: + checkpoint_name (str): The name of the checkpoint. + bucket_name (str): S3 bucket where checkpoint markers are stored. + key_prefix (str): S3 key prefix for checkpoint markers. + inner (NodeHandler): The wrapped NodeHandler for processing nodes. + """ + checkpoint_name: str + bucket_name: str + key_prefix: str + inner: NodeHandler + + def _get_s3_client(self): + from graphrag_toolkit.lexical_graph.config import GraphRAGConfig + return GraphRAGConfig.s3 + + def _marker_key(self, node_id: str) -> str: + return f"{self.key_prefix}/{SAVEPOINT_ROOT_DIR}/{self.checkpoint_name}/{node_id}" + + def touch(self, node_id: str): + """Write a zero-byte S3 object as a checkpoint marker. + + Args: + node_id: The node identifier to use as the marker key. + """ + key = self._marker_key(node_id) + try: + s3_client = self._get_s3_client() + s3_client.put_object(Bucket=self.bucket_name, Key=key, Body=b'') + logger.debug( + f'Checkpoint marker written ' + f'[checkpoint: {self.checkpoint_name}, node_id: {node_id}, ' + f's3://{self.bucket_name}/{key}]' + ) + except Exception as e: + # Non-fatal: worst case is duplicate work on retry + logger.warning( + f'Failed to write checkpoint marker for {node_id}: {e}. ' + f'Node may be re-processed on retry.' + ) + + def accept(self, nodes: List[BaseNode], **kwargs: Any): + """Process nodes via inner handler and write S3 checkpoint markers. + + Args: + nodes: A list of nodes to be processed. + **kwargs: Additional keyword arguments passed to the inner accept method. + + Yields: + BaseNode: Nodes that have been processed by the inner handler. + """ + for node in self.inner.accept(nodes, **kwargs): + node_id = node.node_id + if [key for key in [INDEX_KEY] if key in node.metadata]: + logger.debug( + f'Non-checkpointable node ' + f'[checkpoint: {self.checkpoint_name}, node_id: {node_id}, ' + f'component: {type(self.inner).__name__}]' + ) + else: + logger.debug( + f'Checkpointable node ' + f'[checkpoint: {self.checkpoint_name}, node_id: {node_id}, ' + f'component: {type(self.inner).__name__}]' + ) + self.touch(node_id) + yield node + + +class S3Checkpoint: + """S3-backed checkpoint for data processing components. + + Drop-in replacement for Checkpoint that stores markers as zero-byte S3 objects + instead of local files. Enables checkpointing in serverless/containerized + environments (ECS Fargate, Lambda, EKS) where local disk is ephemeral. + + Marker path: s3://{bucket_name}/{key_prefix}/save_points/{checkpoint_name}/{node_id} + + Usage: + checkpoint = S3Checkpoint( + checkpoint_name='enrichment-run-001', + bucket_name='my-pipeline-bucket', + key_prefix='checkpoints/tenant-a' + ) + graph_index.extract(docs, handler=extracted_docs, checkpoint=checkpoint) + + Attributes: + checkpoint_name (str): The name of the checkpoint. + bucket_name (str): S3 bucket for storing checkpoint markers. + key_prefix (str): S3 key prefix (no trailing slash). + enabled (bool): Whether checkpointing is active. + """ + + def __init__(self, checkpoint_name: str, bucket_name: str, key_prefix: str = '', + region: str = None, enabled: bool = True): + """Initialize an S3-backed checkpoint. + + Args: + checkpoint_name: Name of the checkpoint (used in S3 key path). + bucket_name: S3 bucket where checkpoint markers will be stored. + key_prefix: Optional S3 key prefix (e.g. 'checkpoints/tenant-a'). + No trailing slash needed. + region: AWS region for the S3 bucket. If None, uses default from + GraphRAGConfig or environment. + enabled: Whether checkpointing is active. When False, add_filter + and add_writer return the original objects unwrapped. + """ + self.checkpoint_name = checkpoint_name + self.bucket_name = bucket_name + self.key_prefix = key_prefix.rstrip('/') if key_prefix else '' + self.region = region + self.enabled = enabled + + if self.enabled: + logger.info( + f'S3Checkpoint initialized ' + f'[checkpoint: {checkpoint_name}, ' + f'location: s3://{bucket_name}/{self.key_prefix}/{SAVEPOINT_ROOT_DIR}/{checkpoint_name}/]' + ) + else: + logger.debug( + f'S3Checkpoint disabled [checkpoint: {checkpoint_name}]' + ) + + def add_filter(self, o, tenant_id: TenantId): + """Wrap a TransformComponent with an S3 checkpoint filter. + + Only wraps if enabled, the object is a TransformComponent, and it's not + marked as DoNotCheckpoint. + + Args: + o: The TransformComponent to potentially wrap. + tenant_id: Tenant ID for multi-tenancy node ID rewriting. + + Returns: + S3CheckpointFilter wrapping the input, or the original object. + """ + if self.enabled and isinstance(o, TransformComponent) and not isinstance(o, DoNotCheckpoint): + logger.debug( + f'Wrapping with S3 checkpoint filter ' + f'[checkpoint: {self.checkpoint_name}, component: {type(o).__name__}]' + ) + return S3CheckpointFilter( + inner=o, + bucket_name=self.bucket_name, + key_prefix=self.key_prefix, + checkpoint_name=self.checkpoint_name, + tenant_id=tenant_id, + ) + else: + logger.debug( + f'Not wrapping with S3 checkpoint filter ' + f'[checkpoint: {self.checkpoint_name}, component: {type(o).__name__}]' + ) + return o + + def add_writer(self, o): + """Wrap a NodeHandler with an S3 checkpoint writer. + + Only wraps if enabled and the object is a NodeHandler. + + Args: + o: The NodeHandler to potentially wrap. + + Returns: + S3CheckpointWriter wrapping the input, or the original object. + """ + if self.enabled and isinstance(o, NodeHandler): + logger.debug( + f'Wrapping with S3 checkpoint writer ' + f'[checkpoint: {self.checkpoint_name}, component: {type(o).__name__}]' + ) + return S3CheckpointWriter( + inner=o, + bucket_name=self.bucket_name, + key_prefix=self.key_prefix, + checkpoint_name=self.checkpoint_name, + ) + else: + logger.debug( + f'Not wrapping with S3 checkpoint writer ' + f'[checkpoint: {self.checkpoint_name}, component: {type(o).__name__}]' + ) + return o diff --git a/lexical-graph/tests/unit/indexing/build/test_s3_checkpoint.py b/lexical-graph/tests/unit/indexing/build/test_s3_checkpoint.py new file mode 100644 index 000000000..e95eaefb4 --- /dev/null +++ b/lexical-graph/tests/unit/indexing/build/test_s3_checkpoint.py @@ -0,0 +1,327 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +import pytest +from unittest.mock import Mock, patch, MagicMock +from botocore.exceptions import ClientError + +from graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint import ( + S3Checkpoint, + S3CheckpointFilter, + S3CheckpointWriter, + SAVEPOINT_ROOT_DIR, +) +from graphrag_toolkit.lexical_graph.indexing.build.checkpoint import DoNotCheckpoint +from graphrag_toolkit.lexical_graph.tenant_id import TenantId +from graphrag_toolkit.lexical_graph.storage.constants import INDEX_KEY + + +def _make_404_error(): + """Create a ClientError that simulates S3 404 Not Found.""" + return ClientError( + error_response={'Error': {'Code': '404', 'Message': 'Not Found'}}, + operation_name='HeadObject', + ) + + +def _make_s3_mock(): + """Create a mock S3 client.""" + return MagicMock() + + +class TestS3Checkpoint: + """Tests for S3Checkpoint class.""" + + def test_initialization_enabled(self): + """Verify S3Checkpoint initializes with correct attributes.""" + cp = S3Checkpoint( + checkpoint_name='test-cp', + bucket_name='my-bucket', + key_prefix='prefix/path', + enabled=True, + ) + assert cp.checkpoint_name == 'test-cp' + assert cp.bucket_name == 'my-bucket' + assert cp.key_prefix == 'prefix/path' + assert cp.enabled is True + + def test_initialization_strips_trailing_slash(self): + """Verify key_prefix has trailing slash stripped.""" + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + key_prefix='prefix/', + ) + assert cp.key_prefix == 'prefix' + + def test_initialization_disabled(self): + """Verify S3Checkpoint can be disabled.""" + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + enabled=False, + ) + assert cp.enabled is False + + def test_add_filter_wraps_transform_component(self): + """Verify add_filter wraps a TransformComponent when enabled.""" + from llama_index.core.schema import TransformComponent + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + key_prefix='pfx', + enabled=True, + ) + inner = Mock(spec=TransformComponent) + tenant = TenantId() + result = cp.add_filter(inner, tenant) + assert isinstance(result, S3CheckpointFilter) + + def test_add_filter_skips_do_not_checkpoint(self): + """Verify add_filter does not wrap DoNotCheckpoint instances.""" + from llama_index.core.schema import TransformComponent + + class FakeDoNotCheckpoint(TransformComponent, DoNotCheckpoint): + def __call__(self, nodes, **kwargs): + return nodes + + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + key_prefix='pfx', + enabled=True, + ) + inner = FakeDoNotCheckpoint() + tenant = TenantId() + result = cp.add_filter(inner, tenant) + assert result is inner + + def test_add_filter_disabled(self): + """Verify add_filter returns original when disabled.""" + from llama_index.core.schema import TransformComponent + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + enabled=False, + ) + inner = Mock(spec=TransformComponent) + tenant = TenantId() + result = cp.add_filter(inner, tenant) + assert result is inner + + def test_add_writer_wraps_node_handler(self): + """Verify add_writer wraps a NodeHandler when enabled.""" + from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + key_prefix='pfx', + enabled=True, + ) + inner = Mock(spec=NodeHandler) + result = cp.add_writer(inner) + assert isinstance(result, S3CheckpointWriter) + + def test_add_writer_disabled(self): + """Verify add_writer returns original when disabled.""" + from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler + cp = S3Checkpoint( + checkpoint_name='test', + bucket_name='bucket', + enabled=False, + ) + inner = Mock(spec=NodeHandler) + result = cp.add_writer(inner) + assert result is inner + + +class TestS3CheckpointFilter: + """Tests for S3CheckpointFilter functionality.""" + + def _make_filter(self, s3_mock): + from llama_index.core.schema import TransformComponent + inner = Mock(spec=TransformComponent) + inner.__call__ = Mock(side_effect=lambda nodes, **kw: nodes) + tenant = TenantId() + f = S3CheckpointFilter( + checkpoint_name='test', + bucket_name='my-bucket', + key_prefix='pfx', + inner=inner, + tenant_id=tenant, + ) + return f + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointFilter._get_s3_client') + def test_checkpoint_does_not_exist_returns_true_on_404(self, mock_get_client): + """Verify returns True when S3 HEAD returns 404.""" + s3_mock = _make_s3_mock() + s3_mock.head_object.side_effect = _make_404_error() + mock_get_client.return_value = s3_mock + + f = self._make_filter(s3_mock) + assert f.checkpoint_does_not_exist('node_123') is True + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointFilter._get_s3_client') + def test_checkpoint_does_not_exist_returns_false_when_exists(self, mock_get_client): + """Verify returns False when S3 HEAD succeeds (marker exists).""" + s3_mock = _make_s3_mock() + s3_mock.head_object.return_value = {} # Success = object exists + mock_get_client.return_value = s3_mock + + f = self._make_filter(s3_mock) + assert f.checkpoint_does_not_exist('node_123') is False + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointFilter._get_s3_client') + def test_checkpoint_does_not_exist_returns_true_on_unexpected_error(self, mock_get_client): + """Verify returns True on unexpected errors (safe default: re-process).""" + s3_mock = _make_s3_mock() + s3_mock.head_object.side_effect = ClientError( + error_response={'Error': {'Code': '500', 'Message': 'Internal'}}, + operation_name='HeadObject', + ) + mock_get_client.return_value = s3_mock + + f = self._make_filter(s3_mock) + assert f.checkpoint_does_not_exist('node_123') is True + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointFilter._get_s3_client') + def test_call_filters_checkpointed_nodes(self, mock_get_client): + """Verify __call__ filters out already-checkpointed nodes.""" + s3_mock = _make_s3_mock() + + # n1 exists (checkpointed), n2 does not + def head_side_effect(Bucket, Key): + if 'n1' in Key: + return {} # Exists + raise _make_404_error() + + s3_mock.head_object.side_effect = head_side_effect + mock_get_client.return_value = s3_mock + + from llama_index.core.schema import TransformComponent + inner = Mock(spec=TransformComponent) + inner.__call__ = Mock(side_effect=lambda nodes, **kw: nodes) + tenant = TenantId() + + f = S3CheckpointFilter( + checkpoint_name='test', + bucket_name='my-bucket', + key_prefix='pfx', + inner=inner, + tenant_id=tenant, + ) + + node1 = Mock() + node1.id_ = 'n1' + node2 = Mock() + node2.id_ = 'n2' + + result = f([node1, node2]) + # Only n2 should pass through + assert len(result) == 1 + assert result[0].id_ == 'n2' + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointFilter._get_s3_client') + def test_marker_key_format(self, mock_get_client): + """Verify the S3 key follows expected pattern.""" + s3_mock = _make_s3_mock() + s3_mock.head_object.side_effect = _make_404_error() + mock_get_client.return_value = s3_mock + + f = self._make_filter(s3_mock) + expected = f'pfx/{SAVEPOINT_ROOT_DIR}/test/node_abc' + assert f._marker_key('node_abc') == expected + + +class TestS3CheckpointWriter: + """Tests for S3CheckpointWriter functionality.""" + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointWriter._get_s3_client') + def test_touch_writes_zero_byte_object(self, mock_get_client): + """Verify touch writes a zero-byte S3 object.""" + s3_mock = _make_s3_mock() + mock_get_client.return_value = s3_mock + + from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler + inner = Mock(spec=NodeHandler) + writer = S3CheckpointWriter( + checkpoint_name='test', + bucket_name='my-bucket', + key_prefix='pfx', + inner=inner, + ) + writer.touch('node_xyz') + + s3_mock.put_object.assert_called_once_with( + Bucket='my-bucket', + Key=f'pfx/{SAVEPOINT_ROOT_DIR}/test/node_xyz', + Body=b'', + ) + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointWriter._get_s3_client') + def test_touch_failure_is_non_fatal(self, mock_get_client): + """Verify touch does not raise on S3 errors.""" + s3_mock = _make_s3_mock() + s3_mock.put_object.side_effect = Exception('S3 unavailable') + mock_get_client.return_value = s3_mock + + from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler + inner = Mock(spec=NodeHandler) + writer = S3CheckpointWriter( + checkpoint_name='test', + bucket_name='my-bucket', + key_prefix='pfx', + inner=inner, + ) + # Should not raise + writer.touch('node_xyz') + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointWriter._get_s3_client') + def test_accept_yields_from_inner_and_writes_markers(self, mock_get_client): + """Verify accept yields nodes and writes checkpoint markers.""" + s3_mock = _make_s3_mock() + mock_get_client.return_value = s3_mock + + node = Mock() + node.node_id = 'n1' + node.metadata = {} # No INDEX_KEY = checkpointable + + from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler + inner = Mock(spec=NodeHandler) + inner.accept = Mock(return_value=iter([node])) + + writer = S3CheckpointWriter( + checkpoint_name='test', + bucket_name='my-bucket', + key_prefix='pfx', + inner=inner, + ) + results = list(writer.accept([node])) + assert len(results) == 1 + assert results[0].node_id == 'n1' + s3_mock.put_object.assert_called_once() + + @patch('graphrag_toolkit.lexical_graph.indexing.build.s3_checkpoint.S3CheckpointWriter._get_s3_client') + def test_accept_skips_marker_for_index_nodes(self, mock_get_client): + """Verify accept does not write markers for nodes with INDEX_KEY metadata.""" + s3_mock = _make_s3_mock() + mock_get_client.return_value = s3_mock + + node = Mock() + node.node_id = 'n1' + node.metadata = {INDEX_KEY: {'index': 'chunk', 'key': 'abc'}} + + from graphrag_toolkit.lexical_graph.indexing.node_handler import NodeHandler + inner = Mock(spec=NodeHandler) + inner.accept = Mock(return_value=iter([node])) + + writer = S3CheckpointWriter( + checkpoint_name='test', + bucket_name='my-bucket', + key_prefix='pfx', + inner=inner, + ) + results = list(writer.accept([node])) + assert len(results) == 1 + s3_mock.put_object.assert_not_called()