Tdl 13587 add inventory item data

tap_shopify/schemas/inventory_items.json

@@ -0,0 +1,100 @@
+{
+  "type": "object",
+  "properties": {
+    "id": {
+      "type": [
+        "null",
+        "integer"
+      ]
+    },
+    "sku": {
+      "type": [
+        "null",
+        "string"
+      ]
+    },
+    "created_at": {
+      "type": [
+        "null",
+        "string"
+      ],
+      "format": "date-time"
+    },
+    "updated_at": {
+      "type": [
+        "null",
+        "string"
+      ],
+      "format": "date-time"
+    },
+    "requires_shipping": {
+      "type": [
+        "null",
+        "boolean"
+      ]
+    },
+    "cost": {
+      "type": [
+        "null",
+        "string"
+      ],
+      "format": "singer.decimal"
+    },
+    "country_code_of_origin": {
+      "type": [
+        "null",
+        "string"
+      ]
+    },
+    "province_code_of_origin": {
+      "type": [
+        "null",
+        "string"
+      ]
+    },
+    "harmonized_system_code": {
+      "type": [
+        "null",
+        "integer"
+      ]
+    },
+    "tracked": {
+      "type": [
+        "null",
+        "boolean"
+      ]
+    },
+    "country_harmonized_system_codes": {
+      "type": [
+        "null",
+        "array"
+      ],
+      "items": {
+        "type": [
+          "null",
+          "object"
+        ],
+        "properties": {
+          "harmonized_system_code": {
+            "type": [
+              "null",
+              "string"
+            ]
+          },
+          "country_code": {
+            "type": [
+              "null",
+              "string"
+            ]
+          }
+        }
+      }
+    },
+    "admin_graphql_api_id": {
+      "type": [
+        "null",
+        "string"
+      ]
+    }
+  }
+}

tap_shopify/streams/__init__.py

@@ -7,3 +7,4 @@
 import tap_shopify.streams.products
 import tap_shopify.streams.collects
 import tap_shopify.streams.custom_collections
+import tap_shopify.streams.inventory_items

tap_shopify/streams/inventory_items.py

@@ -0,0 +1,54 @@
+import singer
+import shopify
+from singer.utils import strftime,strptime_to_utc
+from tap_shopify.streams.base import (Stream, shopify_error_handling)
+from tap_shopify.context import Context
+
+LOGGER = singer.get_logger()
+
+RESULTS_PER_PAGE = 250
+
+class InventoryItems(Stream):
+    name = 'inventory_items'
+    replication_object = shopify.InventoryItem
+
+    @shopify_error_handling
+    def get_inventory_items(self, inventory_items_ids):
+        return self.replication_object.find(
+            ids=inventory_items_ids,
+            limit=RESULTS_PER_PAGE)
+
+    def get_objects(self):
+
+        selected_parent = Context.stream_objects['products']()
+        selected_parent.name = "product_variants"
+
+        # Page through all `products`, bookmarking at `product_variants`
+        for parent_object in selected_parent.get_objects():
+
+            product_variants = parent_object.variants
+            inventory_items_ids = ",".join(
+                [str(product_variant.inventory_item_id) for product_variant in product_variants])
+
+            # Max limit of IDs is 100 and Max limit of product_variants in one product is also 100
+            # hence we can directly pass all inventory_items_ids
+            inventory_items = self.get_inventory_items(inventory_items_ids)
+
+            for inventory_item in inventory_items:
+                yield inventory_item
+
+    def sync(self):
+        bookmark = self.get_bookmark()
+        max_bookmark = bookmark
+        for inventory_item in self.get_objects():
+            inventory_item_dict = inventory_item.to_dict()
+            replication_value = strptime_to_utc(inventory_item_dict[self.replication_key])
+            if replication_value >= bookmark:
+                yield inventory_item_dict
+
+            if replication_value > max_bookmark:
+                max_bookmark = replication_value
+
+        self.update_bookmark(strftime(max_bookmark))
+
+Context.stream_objects['inventory_items'] = InventoryItems

tests/base.py

@@ -98,6 +98,10 @@ def expected_metadata(self):
                 self.REPLICATION_METHOD: self.INCREMENTAL,
                 self.API_LIMIT: self.DEFAULT_RESULTS_PER_PAGE},
             "products": default,
+            "inventory_items": {self.REPLICATION_KEYS: {"updated_at"},
+                self.PRIMARY_KEYS: {"id"},
+                self.REPLICATION_METHOD: self.INCREMENTAL,
+                self.API_LIMIT: 250},
             "metafields": meta,
             "transactions": {
                 self.REPLICATION_KEYS: {"created_at"},
@@ -277,5 +281,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','inventory_items'}
+        self.store_2_streams = {'abandoned_checkouts', 'collects', 'metafields', 'transactions', 'order_refunds', 'products','inventory_items'}

tests/test_pagination.py

@@ -18,13 +18,16 @@ def name(self):
 
 
     def test_run(self):
+        # As it can call for max 100 product_variants and 
+        # we can generate only one inventory_item for one product_variants
+        excepted_streams = {'inventory_items'} 
         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):

tests/unittests/test_inventory_items.py

@@ -0,0 +1,92 @@
+import unittest
+from unittest import mock
+from singer.utils import strptime_to_utc
+from tap_shopify.context import Context
+
+INVENTORY_ITEM_OBJECT = Context.stream_objects['inventory_items']()
+
+class Product():
+    def __init__(self, id, variants):
+        self.id = id
+        self.variants = variants
+
+class ProductVariant():
+    def __init__(self, id, inventory_item_id):
+        self.id = id
+        self.inventory_item_id = inventory_item_id
+
+class InventoryItems():
+    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}
+
+ITEM_1 = InventoryItems("i11", "2021-08-11T01:57:05-04:00")
+ITEM_2 = InventoryItems("i12", "2021-08-12T01:57:05-04:00")
+ITEM_3 = InventoryItems("i21", "2021-08-13T01:57:05-04:00")
+ITEM_4 = InventoryItems("i22", "2021-08-14T01:57:05-04:00")
+
+class TestInventoryItems(unittest.TestCase):
+
+    @mock.patch("tap_shopify.streams.base.Stream.get_objects")
+    @mock.patch("tap_shopify.streams.inventory_items.InventoryItems.get_inventory_items")
+    def test_get_objects_with_product_variant(self, mock_get_inventory_items, mock_parent_object):
+
+        expected_inventory_items =  [ITEM_1, ITEM_2, ITEM_3, ITEM_4]
+        product1 = Product("p1", [ProductVariant("v11", "i11"), ProductVariant("v21", "i21")])
+        product2 = Product("p2", [ProductVariant("v12", "i12"), ProductVariant("v22", "i22")])
+
+        mock_get_inventory_items.side_effect = [[ITEM_1, ITEM_2], [ITEM_3, ITEM_4]]
+        mock_parent_object.return_value = [product1, product2]
+
+        actual_inventory_items = list(INVENTORY_ITEM_OBJECT.get_objects())
+
+        #Verify that it returns inventory_item of all product variant
+        self.assertEqual(actual_inventory_items, expected_inventory_items)
+
+
+    @mock.patch("tap_shopify.streams.base.Stream.get_objects")
+    @mock.patch("tap_shopify.streams.inventory_items.InventoryItems.get_inventory_items")
+    def test_get_objects_with_product_but_no_variant(self, mock_get_inventory_items, mock_parent_object):
+
+        expected_inventory_items =  [ITEM_3, ITEM_4]
+
+        #Product1 contain no variant
+        product1 = Product("p1", [])
+
+        product2 = Product("p2", [ProductVariant("v12", "i12"), ProductVariant("v22", "i22")])
+        mock_parent_object.return_value = [product1, product2]
+
+        mock_get_inventory_items.side_effect = [[], [ITEM_3, ITEM_4]] 
+
+        actual_inventory_items = list(INVENTORY_ITEM_OBJECT.get_objects())
+        #Verify that it returns inventory_item of existing product variant
+        self.assertEqual(actual_inventory_items, expected_inventory_items)
+
+
+    @mock.patch("tap_shopify.streams.base.Stream.get_objects")
+    @mock.patch("tap_shopify.streams.inventory_items.InventoryItems.get_inventory_items")
+    def test_get_objects_with_no_product(self, mock_get_inventory_items, mock_parent_object):
+
+        #No product exist
+        mock_parent_object.return_value = []
+        expected_inventory_items = []
+
+        actual_inventory_items = list(INVENTORY_ITEM_OBJECT.get_objects())
+        self.assertEqual(actual_inventory_items, expected_inventory_items)
+
+    @mock.patch("tap_shopify.streams.base.Stream.get_bookmark")
+    @mock.patch("tap_shopify.streams.inventory_items.InventoryItems.get_objects")
+    def test_sync(self, mock_get_objects, mock_get_bookmark):
+
+        expected_sync = [ITEM_3.to_dict(), ITEM_4.to_dict()]
+        mock_get_objects.return_value = [ITEM_1, ITEM_2, ITEM_3, ITEM_4]
+
+        mock_get_bookmark.return_value = strptime_to_utc("2021-08-13T01:05:05-04:00")
+
+        actual_sync = list(INVENTORY_ITEM_OBJECT.sync())
+
+        #Verify that only 2 record syncs
+        self.assertEqual(actual_sync, expected_sync)