New automatic_keys Support and Bulk State Persistence Logic (#237)

* Store the bulk operation details in the state once it is complete for the extraction if the existing fails due to the time limit of 23 hours

* add automatic keys properties

* update the automatic key name

* Increase the chunk size to 1MB

* remove chunk size modifications

* update setup and changelog

* update discovery test

* update automatic test for the order_refunds stream

* include extra automatic keys logic in the automatic test

Author: sgandhi1311

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+### 3.7.2
+  * New automatic_keys Support and Bulk State Persistence Logic [#237](https://github.com/singer-io/tap-shopify/pull/237)
+
 ### 3.7.1
   * Query and extract multiple pages of products for each collection [#227](https://github.com/singer-io/tap-shopify/pull/231)
 

setup.py

@@ -4,7 +4,7 @@
 
 setup(
     name="tap-shopify",
-    version="3.7.1",
+    version="3.7.2",
     description="Singer.io tap for extracting Shopify data",
     author="Stitch",
     url="http://github.com/singer-io/tap-shopify",

tap_shopify/__init__.py

@@ -86,7 +86,8 @@ def get_discovery_metadata(stream, schema):
         mdata = metadata.write(mdata, (), 'valid-replication-keys', [stream.replication_key])
 
     for field_name in schema['properties'].keys():
-        if field_name in stream.key_properties or field_name == stream.replication_key:
+        if field_name in stream.key_properties or field_name == stream.replication_key \
+            or field_name in stream.automatic_keys:
             mdata = metadata.write(mdata, ('properties', field_name), 'inclusion', 'automatic')
         elif field_name in UNSUPPORTED_FIELDS and not has_read_users_access():
             mdata = metadata.write(mdata, ('properties', field_name), 'inclusion', 'unsupported')

tap_shopify/streams/base.py

@@ -132,6 +132,7 @@ class Stream():
     name = None
     replication_method = 'INCREMENTAL'
     replication_key = 'updatedAt'
+    automatic_keys = []
     key_properties = ['id']
     date_window_size = None
     data_key = None

tap_shopify/streams/order_refunds.py

@@ -11,6 +11,7 @@ class OrderRefunds(Stream):
     data_key = "orders"
     child_data_key = "refunds"
     replication_key = "updatedAt"
+    automatic_keys = ["order"]
 
     # pylint: disable=too-many-locals
     def get_objects(self):

tap_shopify/streams/orders.py

@@ -1118,6 +1118,15 @@ def fetch_bulk_operation(op_id):
 
             if current_status == "COMPLETED":
                 LOGGER.info("Bulk operation completed. File size: %s bytes", op.get("fileSize"))
+                self.update_bookmark(
+                    bookmark_value=utils.strftime(current_bookmark),
+                    bulk_op_metadata={
+                        "bulk_operation_id": op.get("id"),
+                        "status": current_status,
+                        "created_at": op.get("createdAt"),
+                        "last_date_window": self.date_window_size,
+                    }
+                )
                 return op.get("url")
 
             if current_status in ["FAILED", "CANCELED"]:
@@ -1281,12 +1290,11 @@ def get_objects(self):
                     current_bookmark = max(current_bookmark, replication_value)
 
                     yield obj
-
-                self.clear_bulk_operation_state()
             else:
                 LOGGER.info("No data returned for the date range: %s to %s",
                                last_updated_at, query_end)
 
+            self.clear_bulk_operation_state()
             last_updated_at = query_end
             max_bookmark_value = min(sync_start, current_bookmark)
             self.update_bookmark(utils.strftime(max_bookmark_value))

tests/test_automatic_fields.py

@@ -64,6 +64,8 @@ def automatic_test(self, conn_id, testable_streams):
                 stream_metadata = self.expected_metadata().get(stream, {})
                 expected_primary_keys = self.expected_primary_keys().get(stream, set())
 
+                extra_automatic_keys = {"order"} if stream == "order_refunds" else set()
+
                 # collect records
                 messages = synced_records.get(stream)
 
@@ -79,7 +81,7 @@ def automatic_test(self, conn_id, testable_streams):
                 # verify that only the automatic fields are sent to the target
                 self.assertEqual(
                     actual_fields_by_stream.get(stream, set()),
-                    expected_primary_keys |
+                    expected_primary_keys | extra_automatic_keys |
                     self.expected_replication_keys().get(stream, set()),
                     msg="The fields sent to the target are not the automatic fields"
                 )

tests/test_discovery.py

@@ -129,6 +129,10 @@ def test_run(self):
                 expected_primary_keys = self.expected_primary_keys()[stream]
                 expected_replication_keys = self.expected_replication_keys()[stream]
                 expected_automatic_fields = expected_primary_keys | expected_replication_keys
+                
+                # In the order_refunds stream, order field is explicitly marked as automatic
+                if stream == "order_refunds":
+                    expected_automatic_fields = expected_automatic_fields | {"order"}
 
                 # verify that primary, replication and foreign keys
                 # are given the inclusion of automatic in annotated schema.