Merge pull request #32 from datazip-inc/feat/start_date_in_log_based

feat: Start Date in log based sync

Author: piyushdatazip

dz_mongodb/__init__.py

@@ -174,6 +174,7 @@ def sync_log_based_streams(client: MongoClient,
                            update_buffer_size: Optional[int],
                            await_time_ms: Optional[int],
                            full_load_on_empty_state: bool,
+                           start_date: Optional[str],
                            document_remove:bool  = False,
                            ):
     """
@@ -209,7 +210,7 @@ def sync_log_based_streams(client: MongoClient,
         update_buffer_size = update_buffer_size or change_streams.MIN_UPDATE_BUFFER_LENGTH
         await_time_ms = await_time_ms or change_streams.DEFAULT_AWAIT_TIME_MS
 
-        change_streams.sync_database(client[database_name], streams, state, update_buffer_size, await_time_ms, full_load_on_empty_state,document_remove)
+        change_streams.sync_database(client[database_name], streams, state, update_buffer_size, await_time_ms, full_load_on_empty_state,start_date,document_remove)
 
     state = singer.set_currently_syncing(state, None)
     singer.write_message(singer.StateMessage(value=copy.deepcopy(state)))
@@ -243,6 +244,7 @@ def do_sync(client: MongoClient, catalog: Dict, config: Dict, state: Dict):
                            config.get('update_buffer_size'),
                            config.get('await_time_ms'),
                            config.get('full_load_on_empty_state'),
+                           config.get('start_date'),
                            config.get('remove_document_prefix')
                            )
     LOGGER.debug('Sync of log based streams done')

dz_mongodb/config_utils.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import Dict
 
 from dz_mongodb.errors import InvalidAwaitTimeError, InvalidLogBasedFullLoadOnEmptyState, InvalidUpdateBufferSizeError
@@ -43,4 +44,9 @@ def validate_config(config: Dict) -> None:
         full_load_on_empty_state = config['full_load_on_empty_state']
 
         if full_load_on_empty_state not in [True, False]:
-            raise InvalidLogBasedFullLoadOnEmptyState(full_load_on_empty_state, 'expected string boolean.')
+            raise InvalidLogBasedFullLoadOnEmptyState(full_load_on_empty_state, 'expected string boolean.')
+        
+    if 'start_date' in config:
+        start_datetime = datetime.strptime(config['start_date'], "%Y-%m-%d")
+        if not isinstance(start_datetime, datetime):
+            raise ValueError('start_date must be a datetime object.')

dz_mongodb/sync_strategies/change_streams.py

@@ -1,5 +1,7 @@
 import copy
+from datetime import datetime
 import time
+from bson import ObjectId
 import singer
 import pymongo
 
@@ -77,6 +79,7 @@ def sync_database(database: Database,
                   update_buffer_size: int,
                   await_time_ms: int,
                   full_load_on_empty_state: bool,
+                  start_date: Optional[str],
                   document_remove: bool = False
                   ) -> None:
     """
@@ -111,9 +114,13 @@ def sync_database(database: Database,
         for tap_stream_id in full_load:
             table_name = streams_to_sync[tap_stream_id].get('table_name')
             collection = database[table_name]
-
+            filter = {}
+            if start_date:
+                start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
+                filter = { "_id": { "$gte": ObjectId.from_datetime(start_datetime) }}
+                LOGGER.info('using filter for date[%s] to fetch data: %s',start_datetime, filter)
             # TODO: add batches
-            with collection.find(sort=[("_id", pymongo.ASCENDING)]) as cursor:
+            with collection.find(filter,sort=[("_id", pymongo.ASCENDING)]) as cursor:
                 for row in cursor:
                     rows_saved[tap_stream_id] += 1
                     singer.write_message(common.row_to_singer_record(stream=streams_to_sync[tap_stream_id],

setup.py

@@ -5,7 +5,7 @@
     long_desc = fh.read()
 
 setup(name='dz-mongodb',
-      version='1.4.2',
+      version='1.4.3',
       description='Singer.io tap for extracting data from MongoDB - Datazip compatible',
       long_description=long_desc,
       long_description_content_type='text/markdown',