From d997cf0c84a4137d1fcaea164244c5c584edbac2 Mon Sep 17 00:00:00 2001 From: Serhii Lazebnyi Date: Tue, 17 Dec 2024 05:18:52 +0100 Subject: [PATCH] Fix checkpointing for declarative streams --- airbyte_cdk/sources/streams/core.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/streams/core.py b/airbyte_cdk/sources/streams/core.py index a9aa8550..6cc5c8b5 100644 --- a/airbyte_cdk/sources/streams/core.py +++ b/airbyte_cdk/sources/streams/core.py @@ -223,17 +223,17 @@ def read( # type: ignore # ignoring typing for ConnectorStateManager because o record_counter += 1 checkpoint_interval = self.state_checkpoint_interval - checkpoint = checkpoint_reader.get_checkpoint() if ( should_checkpoint and checkpoint_interval and record_counter % checkpoint_interval == 0 - and checkpoint is not None ): - airbyte_state_message = self._checkpoint_state( - checkpoint, state_manager=state_manager - ) - yield airbyte_state_message + checkpoint = checkpoint_reader.get_checkpoint() + if checkpoint: + airbyte_state_message = self._checkpoint_state( + checkpoint, state_manager=state_manager + ) + yield airbyte_state_message if internal_config.is_limit_reached(record_counter): break