add pagination for fulfillment line items

Author: sgandhi1311

tap_shopify/streams/fulfillment_orders.py

@@ -65,6 +65,72 @@ def transform_childitems(self, data, parent_id, key, next_page_key):
 
         return child_records
 
+    def get_fulfillment_line_items(self, fulfillment_id, next_page=None):
+        """
+        Fetch all fulfillment line items for a given fulfillment ID.
+        """
+        fulfillment_line_items = []
+        query = """
+                query FulfillmentLineItems($fulfillmentId: ID!, $next_page: String) {
+                fulfillment(id: $fulfillmentId) {
+                    fulfillmentLineItems(first: 1, after: $next_page) {
+                    pageInfo {
+                        hasNextPage
+                        endCursor
+                    }
+                    nodes {
+                        id
+                        quantity
+                        originalTotalSet {
+                        presentmentMoney {
+                            amount
+                            currencyCode
+                        }
+                        shopMoney {
+                            amount
+                            currencyCode
+                        }
+                        }
+                        discountedTotalSet {
+                        presentmentMoney {
+                            amount
+                            currencyCode
+                        }
+                        shopMoney {
+                            amount
+                            currencyCode
+                        }
+                        }
+                        lineItem {
+                        id
+                        }
+                    }
+                    }
+                }
+                }
+                """
+
+        params = {
+            "fulfillmentId": fulfillment_id,
+        }
+        if next_page:
+            params["next_page"] = next_page
+
+        while True:
+            response = self.call_api(params, query=query, data_key="fulfillment")
+            fulfillment_line_items.extend(
+                node for item in response.get("fulfillmentLineItems", {}).get("nodes", [])
+                if (node := item)
+            )
+
+            page_info = response.get("fulfillmentLineItems", {}).get("pageInfo", {})
+            if not page_info.get("hasNextPage"):
+                break
+
+            params["next_page"] = page_info.get("endCursor")
+
+        return fulfillment_line_items
+
     def transform_object(self, obj):
         """
         Transforms a collection object.
@@ -98,8 +164,16 @@ def transform_object(self, obj):
             )
             for item in obj["fulfillments"]:
                 item["fulfillmentOrders"] = item["fulfillmentOrders"]["nodes"]
-                item["fulfillmentLineItems"] = item["fulfillmentLineItems"]["nodes"]
                 item["events"] = item["events"]["nodes"]
+                initial_nodes = item["fulfillmentLineItems"]["nodes"]
+                if item["fulfillmentLineItems"]["pageInfo"]["hasNextPage"]:
+                    more_nodes = self.get_fulfillment_line_items(
+                        fulfillment_id=item["id"],
+                        next_page = item["fulfillmentLineItems"]["pageInfo"]["endCursor"]
+                    )
+                    item["fulfillmentLineItems"] = more_nodes + initial_nodes
+                else:
+                    item["fulfillmentLineItems"] = initial_nodes
 
         if obj.get("fulfillmentOrdersForMerge"):
             obj["fulfillmentOrdersForMerge"] = obj["fulfillmentOrdersForMerge"]["nodes"]
@@ -286,6 +360,10 @@ def get_query(self):
                                             }
                                         }
                                         fulfillmentLineItems(first: 3) {
+                                            pageInfo {
+                                                    hasNextPage
+                                                    endCursor
+                                                }
                                             nodes {
                                                 id
                                                 quantity
@@ -313,7 +391,6 @@ def get_query(self):
                                                     id
                                                 }
                                             }
-                                            
                                         }
                                         legacyResourceId
                                         order {