Add first state msg in Read method

airbytehq · Sep 27, 2022 · f13d6c5 · f13d6c5
1 parent 619bdb9
commit f13d6c5
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/airbyte-integrations/connectors/source-dv-360/source_dv_360/source.py b/airbyte-integrations/connectors/source-dv-360/source_dv_360/source.py
@@ -7,7 +7,7 @@
 from airbyte_cdk.logger import AirbyteLogger
 from airbyte_cdk.models import (
     AirbyteCatalog,
-    AirbyteConnectionStatus,
+    AirbyteStateMessage,
     AirbyteMessage,
     AirbyteRecordMessage,
     AirbyteStream,
@@ -121,9 +121,10 @@ def read(self, logger: AirbyteLogger, config: json, catalog: ConfiguredAirbyteCa
             stream_state= state.get(stream_name, {})
             #if stream_state and "state" in dir(stream_instance):
             stream_instance.state= stream_state
+            logger.info(f"Syncing {stream_name} stream")
             logger.info(f"Setting state of {stream_name} stream to {stream_state}")
+            yield AirbyteMessage(type=Type.STATE, state=AirbyteStateMessage(data=state))
             try:
-                logger.info(f"Syncing {stream_name} stream")
                 config_catalog_fields= configured_stream.stream.json_schema.get('properties').keys()
                 slices= stream_instance.stream_slices(
                     cursor_field= configured_stream.cursor_field,

diff --git a/airbyte-integrations/connectors/source-dv-360/source_dv_360/streams.py b/airbyte-integrations/connectors/source-dv-360/source_dv_360/streams.py
@@ -258,7 +258,7 @@ def buffer_reader(self, buffer:io.StringIO):
 class DBMIncrementalStream(DBMStream, ABC):
   cursor_field = "date"
   primary_key = None
-  range_days = 15
+  range_days = 30 #range of stream slice
 
   def __init__(self, credentials: Credentials, partner_id: str, filters:List[dict], start_date: str, end_date: str=None):
     super().__init__(credentials, partner_id, filters, start_date, end_date)