🎉 Source Shopify: Add BalanceTransactions stream

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/9da77001-af33-4bcd-be46-6252bf9342b9.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "9da77001-af33-4bcd-be46-6252bf9342b9",
   "name": "Shopify",
   "dockerRepository": "airbyte/source-shopify",
-  "dockerImageTag": "0.1.30",
+  "dockerImageTag": "0.1.31",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/shopify",
   "icon": "shopify.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -655,7 +655,7 @@
 - name: Shopify
   sourceDefinitionId: 9da77001-af33-4bcd-be46-6252bf9342b9
   dockerRepository: airbyte/source-shopify
-  dockerImageTag: 0.1.30
+  dockerImageTag: 0.1.31
   documentationUrl: https://docs.airbyte.io/integrations/sources/shopify
   icon: shopify.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -6994,7 +6994,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-shopify:0.1.30"
+- dockerImage: "airbyte/source-shopify:0.1.31"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/shopify"
     connectionSpecification:

airbyte-integrations/connectors/source-shopify/Dockerfile

@@ -28,5 +28,5 @@ COPY source_shopify ./source_shopify
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.30
+LABEL io.airbyte.version=0.1.31
 LABEL io.airbyte.name=airbyte/source-shopify

airbyte-integrations/connectors/source-shopify/integration_tests/abnormal_state.json

@@ -55,5 +55,8 @@
   },
   "fulfillments": {
     "updated_at": "2024-07-08T05:40:38-07:00"
+  },
+  "balance_transactions": {
+    "id": 9999999999999
   }
 }

airbyte-integrations/connectors/source-shopify/integration_tests/configured_catalog.json

@@ -237,6 +237,18 @@
       "sync_mode": "incremental",
       "cursor_field": ["updated_at"],
       "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "balance_transactions",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["id"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["id"],
+      "destination_sync_mode": "append"
     }
   ]
 }

airbyte-integrations/connectors/source-shopify/integration_tests/state.json

@@ -55,5 +55,8 @@
   },
   "fulfillments": {
     "updated_at": "2021-09-10T06:48:10-07:00"
+  },
+  "balance_transactions": {
+    "id": 29427031703741
   }
 }

airbyte-integrations/connectors/source-shopify/source_shopify/schemas/balance_transactions.json

@@ -0,0 +1,52 @@
+{
+  "type": ["null", "object"],
+  "properties": {
+    "id": {
+      "type": "integer"
+    },
+    "type": {
+      "type": ["null", "string"]
+    },
+    "test": {
+      "type": ["null", "boolean"]
+    },
+    "payout_id": {
+      "type": ["null", "integer"]
+    },
+    "payout_status": {
+      "type": ["null", "string"]
+    },
+    "payoucurrencyt_status": {
+      "type": ["null", "string"]
+    },
+    "amount": {
+      "type": ["null", "number"]
+    },
+    "fee": {
+      "type": ["null", "number"]
+    },
+    "net": {
+      "type": ["null", "number"]
+    },
+    "source_id": {
+      "type": ["null", "integer"]
+    },
+    "source_type": {
+      "type": ["null", "string"]
+    },
+    "source_order_transaction_id": {
+      "type": ["null", "integer"]
+    },
+    "source_order_id": {
+      "type": ["null", "integer"]
+    },
+    "processed_at": {
+      "type": ["null", "string"],
+      "format": "date-time"
+    },
+    "shop_url": {
+      "type": ["null", "string"]
+    }
+  }
+
+}

airbyte-integrations/connectors/source-shopify/source_shopify/source.py

@@ -249,25 +249,12 @@ def path(self, **kwargs) -> str:
         return f"{self.data_field}.json"
 
 
-class Collects(IncrementalShopifyStream):
-    """
-    Collects stream does not support Incremental Refresh based on datetime fields, only `since_id` is supported:
-    https://shopify.dev/docs/admin-api/rest/reference/products/collect
-
-    The Collect stream is the link between Products and Collections, if the Collection is created for Products,
-    the `collect` record is created, it's reasonable to Full Refresh all collects. As for Incremental refresh -
-    we would use the since_id specificaly for this stream.
+class IncrementalByIDShopifyStream(IncrementalShopifyStream):
 
-    """
-
-    data_field = "collects"
     cursor_field = "id"
     order_field = "id"
     filter_field = "since_id"
 
-    def path(self, **kwargs) -> str:
-        return f"{self.data_field}.json"
-
     def get_updated_state(self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]) -> Mapping[str, Any]:
         return {self.cursor_field: max(latest_record.get(self.cursor_field, 0), current_stream_state.get(self.cursor_field, 0))}
 
@@ -281,6 +268,35 @@ def request_params(
         return params
 
 
+class Collects(IncrementalByIDShopifyStream):
+    """
+    Collects stream does not support Incremental Refresh based on datetime fields, only `since_id` is supported:
+    https://shopify.dev/docs/admin-api/rest/reference/products/collect
+
+    The Collect stream is the link between Products and Collections, if the Collection is created for Products,
+    the `collect` record is created, it's reasonable to Full Refresh all collects. As for Incremental refresh -
+    we would use the since_id specificaly for this stream.
+    """
+
+    data_field = "collects"
+
+    def path(self, **kwargs) -> str:
+        return f"{self.data_field}.json"
+
+
+class BalanceTransactions(IncrementalByIDShopifyStream):
+
+    """
+    PaymentsTransactions stream does not support Incremental Refresh based on datetime fields, only `since_id` is supported:
+    https://shopify.dev/api/admin-rest/2021-07/resources/transactions
+    """
+
+    data_field = "transactions"
+
+    def path(self, **kwargs) -> str:
+        return f"shopify_payments/balance/{self.data_field}.json"
+
+
 class OrderRefunds(ChildSubstream):
     parent_stream_class: object = Orders
     slice_key = "order_id"
@@ -474,6 +490,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             OrderRefunds(config),
             OrderRisks(config),
             Transactions(config),
+            BalanceTransactions(config),
             Pages(config),
             PriceRules(config),
             DiscountCodes(config),

airbyte-integrations/connectors/source-shopify/source_shopify/utils.py

@@ -20,6 +20,7 @@
     "read_locations": ["Locations"],
     "read_inventory": ["InventoryItems", "InventoryLevels"],
     "read_merchant_managed_fulfillment_orders": ["FulfillmentOrders"],
+    "read_shopify_payments_payouts": ["BalanceTransactions"],
 }
 
 

docs/integrations/sources/shopify.md

@@ -33,6 +33,7 @@ This Source is capable of syncing the following core Streams:
 * [Orders Risks](https://shopify.dev/api/admin/rest/reference/orders/order-risk)
 * [Products](https://help.shopify.com/en/api/reference/products)
 * [Transactions](https://help.shopify.com/en/api/reference/orders/transaction)
+* [Balance Transactions](https://shopify.dev/api/admin-rest/2021-07/resources/transactions)
 * [Pages](https://help.shopify.com/en/api/reference/online-store/page)
 * [Price Rules](https://help.shopify.com/en/api/reference/discounts/pricerule)
 * [Locations](https://shopify.dev/api/admin-rest/2021-10/resources/location)
@@ -101,6 +102,7 @@ This connector support both: `OAuth 2.0` and `API PASSWORD` (for private applica
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| 0.1.31 | 2021-01-26 | [9850](https://github.com/airbytehq/airbyte/pull/9850) | Added `BalanceTransactions` |
 | 0.1.30 | 2021-01-24 | [9648](https://github.com/airbytehq/airbyte/pull/9648) | Added permission validation before sync |
 | 0.1.29 | 2022-01-20 | [9049](https://github.com/airbytehq/airbyte/pull/9248) | Added `shop_url` to the record for all streams |
 | 0.1.28 | 2022-01-19 | [9591](https://github.com/airbytehq/airbyte/pull/9591) | Implemented `OAuth2.0` authentication method for Airbyte Cloud |