TDL-13615: Add Locations stream and TDL-13614: Add Inventory Levels stream

README.md

@@ -16,6 +16,8 @@ This tap:
   - [Orders](https://help.shopify.com/en/api/reference/orders)
   - [Products](https://help.shopify.com/en/api/reference/products)
   - [Transactions](https://help.shopify.com/en/api/reference/orders/transaction)
+  - [Locations](https://help.shopify.com/en/api/reference/inventory/location)
+  - [Inventory Levels](https://help.shopify.com/en/api/reference/inventory/inventorylevel)
 - Outputs the schema for each resource
 - Incrementally pulls data based on the input state
 - When Metafields are selected, this tap will sync the Shopify store's top-level Metafields and any additional Metafields for selected tables that also have them (ie: Orders, Products, Customers)

tap_shopify/__init__.py

@@ -156,6 +156,11 @@ def sync():
             except pyactiveresource.connection.UnauthorizedAccess as exc:
                 raise ShopifyError(exc, 'Invalid access token - Re-authorize the connection') \
                     from exc
+            except pyactiveresource.connection.ForbiddenAccess as exc:
+                raise ShopifyError(exc, 'Add read_inventory scope for access token and '
+                                   'Re-authorize the connection to sync '
+                                   'Inventory Levels and Inventory Items streams.') \
+                    from exc
             except pyactiveresource.connection.ConnectionError as exc:
                 msg = ''
                 try:

tap_shopify/schemas/definitions.json

@@ -476,6 +476,70 @@
           "null",
           "string"
         ]
+      },
+      "localized_province_name": {
+        "type": [
+          "null",
+          "string"
+        ]
+      },
+      "localized_country_name": {
+        "type": [
+          "null",
+          "string"
+        ]
+      },
+      "updated_at": {
+        "type": [
+          "null",
+          "string"
+        ],
+        "format": "date-time"
+      },
+      "province": {
+        "type": [
+          "null",
+          "string"
+        ]
+      },
+      "phone": {
+        "type": [
+          "null",
+          "string"
+        ]
+      },
+      "legacy": {
+        "type": [
+          "null",
+          "boolean"]
+      },
+      "created_at": {
+        "type": [
+          "null",
+          "string"
+        ],
+        "format": "date-time"
+      },
+      "country": {
+        "type": [
+          "null",
+          "string"
+        ]
+      },
+      "active": {
+        "type": [
+          "null",
+          "boolean"]
+      },
+      "admin_graphql_api_id": {
+        "type": [
+          "null",
+          "string"]
+      },
+      "country_name": {
+        "type": [
+          "null",
+          "string"]
       }
     },
     "type": [

tap_shopify/schemas/inventory_levels.json

@@ -0,0 +1,21 @@
+{
+  "properties": {
+    "available": {
+      "type": ["null", "integer"]
+    },
+    "inventory_item_id": {
+      "type": ["null", "integer"]
+    },
+    "updated_at": {
+      "type": ["null", "string"],
+      "format": "date-time"
+    },
+    "location_id": {
+      "type": ["null", "integer"]
+    },
+    "admin_graphql_api_id": {
+      "type": ["null", "string"]
+    }
+  },
+  "type": "object"
+}

tap_shopify/schemas/locations.json

@@ -0,0 +1,3 @@
+{
+  "$ref": "definitions.json#/location"
+}

tap_shopify/streams/__init__.py

@@ -7,3 +7,5 @@
 import tap_shopify.streams.products
 import tap_shopify.streams.collects
 import tap_shopify.streams.custom_collections
+import tap_shopify.streams.locations
+import tap_shopify.streams.inventory_levels

tap_shopify/streams/inventory_levels.py

@@ -0,0 +1,59 @@
+import shopify
+from singer.utils import strftime, strptime_to_utc
+from tap_shopify.streams.base import (Stream,
+                                      RESULTS_PER_PAGE,
+                                      shopify_error_handling)
+from tap_shopify.context import Context
+
+class InventoryLevels(Stream):
+    name = 'inventory_levels'
+    replication_key = 'updated_at'
+    key_properties = ['location_id', 'inventory_item_id']
+    replication_object = shopify.InventoryLevel
+
+    @shopify_error_handling
+    def api_call_for_inventory_levels(self, parent_object_id, bookmark):
+        return self.replication_object.find(
+            updated_at_min = bookmark,
+            limit = RESULTS_PER_PAGE,
+            location_ids=parent_object_id
+        )
+
+    def get_inventory_levels(self, parent_object, bookmark):
+        inventory_page = self.api_call_for_inventory_levels(parent_object, bookmark)
+        yield from inventory_page
+
+        while inventory_page.has_next_page():
+            inventory_page = inventory_page.next_page()
+            yield from inventory_page
+
+    def get_objects(self):
+        bookmark = self.get_bookmark()
+
+        selected_parent = Context.stream_objects['locations']()
+        selected_parent.name = "inventory_level_locations"
+
+        # Get all locations data as location id is used for Inventory Level
+        # If we get locations updated after a bookmark
+        # then there is possibility of data loss for Inventory Level
+        # because location is not updated when any Inventory Level is updated inside it.
+        for parent_object in selected_parent.get_locations_data():
+            inventory_levels = self.get_inventory_levels(parent_object.id, bookmark)
+            for inventory_level in inventory_levels:
+                yield inventory_level
+
+    def sync(self):
+        bookmark = self.get_bookmark()
+        max_bookmark = bookmark
+        for inventory_level in self.get_objects():
+            inventory_level_dict = inventory_level.to_dict()
+            replication_value = strptime_to_utc(inventory_level_dict[self.replication_key])
+            if replication_value >= bookmark:
+                yield inventory_level_dict
+
+            if replication_value > max_bookmark:
+                max_bookmark = replication_value
+
+        self.update_bookmark(strftime(max_bookmark))
+
+Context.stream_objects['inventory_levels'] = InventoryLevels

tap_shopify/streams/locations.py

@@ -0,0 +1,37 @@
+import shopify
+from singer import utils
+from tap_shopify.streams.base import (Stream, shopify_error_handling)
+from tap_shopify.context import Context
+
+class Locations(Stream):
+    name = 'locations'
+    replication_object = shopify.Location
+
+    @shopify_error_handling
+    def get_locations_data(self):
+        location_page = self.replication_object.find()
+        yield from location_page
+
+        while location_page.has_next_page():
+            location_page = location_page.next_page()
+            yield from location_page
+
+    def sync(self):
+        bookmark = self.get_bookmark()
+        max_bookmark = bookmark
+
+        for location in self.get_locations_data():
+
+            location_dict = location.to_dict()
+            replication_value = utils.strptime_to_utc(location_dict[self.replication_key])
+
+            if replication_value >= bookmark:
+                yield location_dict
+
+            # update max bookmark if "replication_value" of current location is greater
+            if replication_value > max_bookmark:
+                max_bookmark = replication_value
+
+        self.update_bookmark(utils.strftime(max_bookmark))
+
+Context.stream_objects['locations'] = Locations

tests/base.py

@@ -104,6 +104,16 @@ def expected_metadata(self):
                 self.PRIMARY_KEYS: {"id"},
                 self.FOREIGN_KEYS: {"order_id"},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
+                self.API_LIMIT: self.DEFAULT_RESULTS_PER_PAGE},
+            "locations": {
+                self.REPLICATION_KEYS: {"updated_at"},
+                self.PRIMARY_KEYS: {"id"},
+                self.REPLICATION_METHOD: self.INCREMENTAL,
+                self.API_LIMIT: 0},
+            "inventory_levels": {
+                self.REPLICATION_KEYS: {"updated_at"},
+                self.PRIMARY_KEYS: {"location_id", "inventory_item_id"},
+                self.REPLICATION_METHOD: self.INCREMENTAL,
                 self.API_LIMIT: self.DEFAULT_RESULTS_PER_PAGE}
         }
 
@@ -277,5 +287,5 @@ def select_all_streams_and_fields(conn_id, catalogs, select_all_fields: bool = T
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.start_date = self.get_properties().get("start_date")
-        self.store_1_streams = {'custom_collections', 'orders', 'products', 'customers'}
-        self.store_2_streams = {'abandoned_checkouts', 'collects', 'metafields', 'transactions', 'order_refunds', 'products'}
+        self.store_1_streams = {'custom_collections', 'orders', 'products', 'customers', 'locations', 'inventory_levels'}
+        self.store_2_streams = {'abandoned_checkouts', 'collects', 'metafields', 'transactions', 'order_refunds', 'products', 'locations', 'inventory_levels'}

tests/test_pagination.py

@@ -18,13 +18,17 @@ def name(self):
 
 
     def test_run(self):
+        # skip 'locations' stream as there is not much info about
+        # limit of records returned in 1 page
+        # Documentation: https://help.shopify.com/en/manual/locations/setting-up-your-locations
+        excepted_streams = {'locations'}
         with self.subTest(store="store_1"):
             conn_id = self.create_connection(original_credentials=True)
-            self.pagination_test(conn_id, self.store_1_streams)
+            self.pagination_test(conn_id, self.store_1_streams - excepted_streams)
 
         with self.subTest(store="store_2"):
             conn_id = self.create_connection(original_properties=False, original_credentials=False)
-            self.pagination_test(conn_id, self.store_2_streams)
+            self.pagination_test(conn_id, self.store_2_streams - excepted_streams)
 
 
     def pagination_test(self, conn_id, testable_streams):
@@ -55,6 +59,7 @@ def pagination_test(self, conn_id, testable_streams):
         # Run a sync job using orchestrator
         record_count_by_stream = self.run_sync(conn_id)
         actual_fields_by_stream = runner.examine_target_output_for_fields()
+        sync_records = runner.get_records_from_target_output()
 
         for stream in testable_streams:
             with self.subTest(stream=stream):
@@ -70,10 +75,13 @@ def pagination_test(self, conn_id, testable_streams):
                     minimum_record_count,
                     msg="The number of records is not over the stream max limit")
 
+                expected_pk = self.expected_primary_keys()
+                sync_messages = sync_records.get(stream, {'messages': []}).get('messages')
+
                 # verify that the automatic fields are sent to the target
                 self.assertTrue(
                     actual_fields_by_stream.get(stream, set()).issuperset(
-                        self.expected_primary_keys().get(stream, set()) |
+                        expected_pk.get(stream, set()) |
                         self.expected_replication_keys().get(stream, set()) |
                         self.expected_foreign_keys().get(stream, set())),
                     msg="The fields sent to the target don't include all automatic fields"
@@ -83,8 +91,18 @@ def pagination_test(self, conn_id, testable_streams):
                 # SKIP THIS ASSERTION IF ALL FIELDS ARE INTENTIONALLY AUTOMATIC FOR THIS STREAM
                 self.assertTrue(
                     actual_fields_by_stream.get(stream, set()).symmetric_difference(
-                        self.expected_primary_keys().get(stream, set()) |
+                        expected_pk.get(stream, set()) |
                         self.expected_replication_keys().get(stream, set()) |
                         self.expected_foreign_keys().get(stream, set())),
                     msg="The fields sent to the target don't include non-automatic fields"
                 )
+
+                # Verify we did not duplicate any records across pages
+                records_pks_set = {tuple([message.get('data').get(primary_key)
+                                          for primary_key in expected_pk.get(stream, set())])
+                                   for message in sync_messages}
+                records_pks_list = [tuple([message.get('data').get(primary_key)
+                                           for primary_key in expected_pk.get(stream, set())])
+                                    for message in sync_messages]
+                self.assertCountEqual(records_pks_set, records_pks_list,
+                                      msg=f"We have duplicate records for {stream}")

tests/unittests/test_inventory_levels.py

@@ -0,0 +1,76 @@
+import unittest
+from unittest import mock
+from singer.utils import strptime_to_utc
+from tap_shopify.context import Context
+
+INVENTORY_LEVEL_OBJECT = Context.stream_objects['inventory_levels']()
+
+class Location():
+    def __init__(self, id):
+        self.id = id
+
+class InventoryLevels():
+    def __init__(self, id, updated_at):
+        self.id = id
+        self.updated_at = updated_at
+
+    def to_dict(self):
+        return {"id": self.id, "updated_at": self.updated_at}
+
+LEVEL_1 = InventoryLevels("inv_level1", "2021-08-11T01:57:05-04:00")
+LEVEL_2 = InventoryLevels("inv_level2", "2021-08-12T01:57:05-04:00")
+LEVEL_3 = InventoryLevels("inv_level3", "2021-08-13T01:57:05-04:00")
+LEVEL_4 = InventoryLevels("inv_level4", "2021-08-14T01:57:05-04:00")
+
+@mock.patch("tap_shopify.streams.base.Stream.get_bookmark")
+class TestInventoryItems(unittest.TestCase):
+
+    @mock.patch("tap_shopify.streams.locations.Locations.get_locations_data")
+    @mock.patch("tap_shopify.streams.inventory_levels.InventoryLevels.get_inventory_levels")
+    def test_get_objects_with_locations(self, mock_get_inventory_levels, mock_parent_object, mock_get_bookmark):
+        '''
+            Verify that expected data should be emitted for inventory_levels if locations found.
+        '''
+        expected_inventory_levels =  [LEVEL_1, LEVEL_2, LEVEL_3, LEVEL_4]
+        location1 = Location("location1")
+        location2 = Location("location2")
+
+        mock_get_inventory_levels.side_effect = [[LEVEL_1, LEVEL_2], [LEVEL_3, LEVEL_4]]
+        mock_parent_object.return_value = [location1, location2]
+
+        actual_inventory_levels = list(INVENTORY_LEVEL_OBJECT.get_objects())
+
+        #Verify that it returns inventory_levels for all locations
+        self.assertEqual(actual_inventory_levels, expected_inventory_levels)
+
+    @mock.patch("tap_shopify.streams.locations.Locations.get_locations_data")
+    @mock.patch("tap_shopify.streams.inventory_levels.InventoryLevels.get_inventory_levels")
+    def test_get_objects_with_no_locations(self, mock_get_inventory_levels, mock_parent_object, mock_get_bookmark):
+        '''
+            Verify that no data should be emitted for inventory_levels if no locations found.
+        '''
+        # No data for parent stream location
+        mock_parent_object.return_value = []
+        expected_inventory_levels = []
+
+        actual_inventory_levels = list(INVENTORY_LEVEL_OBJECT.get_objects())
+
+        # No get_inventory_levels should be called and no data should be returned 
+        self.assertEqual(actual_inventory_levels, expected_inventory_levels)
+        self.assertEqual(mock_get_inventory_levels.call_count, 0)
+
+    @mock.patch("tap_shopify.streams.inventory_levels.InventoryLevels.get_objects")
+    def test_sync(self, mock_get_objects, mock_get_bookmark):
+        '''
+            Verify that only data updated after specific bookmark are yielded from sync.
+        '''
+
+        expected_sync = [LEVEL_3.to_dict(), LEVEL_4.to_dict()]
+        mock_get_objects.return_value = [LEVEL_1, LEVEL_2, LEVEL_3, LEVEL_4]
+
+        mock_get_bookmark.return_value = strptime_to_utc("2021-08-13T01:05:05-04:00")
+
+        actual_sync = list(INVENTORY_LEVEL_OBJECT.sync())
+
+        #Verify that only 2 record syncs
+        self.assertEqual(actual_sync, expected_sync)