🎉 Source Stripe: use start_date parameter in incremental streams (#…

…6466)

* Use `start_date` parameter in incremental streams

* Bump connector's version + update docs

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e094cb9a-26de-4645-8761-65c0c425d1de.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "e094cb9a-26de-4645-8761-65c0c425d1de",
   "name": "Stripe",
   "dockerRepository": "airbyte/source-stripe",
-  "dockerImageTag": "0.1.18",
+  "dockerImageTag": "0.1.19",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/stripe",
   "icon": "stripe.svg"
 }

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

@@ -126,7 +126,7 @@
 - sourceDefinitionId: e094cb9a-26de-4645-8761-65c0c425d1de
   name: Stripe
   dockerRepository: airbyte/source-stripe
-  dockerImageTag: 0.1.18
+  dockerImageTag: 0.1.19
   documentationUrl: https://docs.airbyte.io/integrations/sources/stripe
   icon: stripe.svg
   sourceType: api

airbyte-integrations/connectors/source-stripe/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.18
+LABEL io.airbyte.version=0.1.19
 LABEL io.airbyte.name=airbyte/source-stripe

airbyte-integrations/connectors/source-stripe/setup.py

@@ -5,7 +5,7 @@
 
 from setuptools import find_packages, setup
 
-MAIN_REQUIREMENTS = ["airbyte-cdk~=0.1", "stripe==2.56.0"]
+MAIN_REQUIREMENTS = ["airbyte-cdk~=0.1", "stripe==2.56.0", "pendulum==1.2.0"]
 
 TEST_REQUIREMENTS = [
     "pytest~=6.1",

airbyte-integrations/connectors/source-stripe/source_stripe/source.py

@@ -45,24 +45,25 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         authenticator = TokenAuthenticator(config["client_secret"])
         args = {"authenticator": authenticator, "account_id": config["account_id"]}
+        incremental_args = {**args, "start_date": config["start_date"]}
         return [
-            BalanceTransactions(**args),
+            BalanceTransactions(**incremental_args),
             BankAccounts(**args),
-            Charges(**args),
-            Coupons(**args),
+            Charges(**incremental_args),
+            Coupons(**incremental_args),
             CustomerBalanceTransactions(**args),
-            Customers(**args),
-            Disputes(**args),
-            Events(**args),
-            InvoiceItems(**args),
+            Customers(**incremental_args),
+            Disputes(**incremental_args),
+            Events(**incremental_args),
+            InvoiceItems(**incremental_args),
             InvoiceLineItems(**args),
-            Invoices(**args),
-            PaymentIntents(**args),
-            Payouts(**args),
-            Plans(**args),
-            Products(**args),
-            Refunds(**args),
+            Invoices(**incremental_args),
+            PaymentIntents(**incremental_args),
+            Payouts(**incremental_args),
+            Plans(**incremental_args),
+            Products(**incremental_args),
+            Refunds(**incremental_args),
             SubscriptionItems(**args),
-            Subscriptions(**args),
-            Transfers(**args),
+            Subscriptions(**incremental_args),
+            Transfers(**incremental_args),
         ]

airbyte-integrations/connectors/source-stripe/source_stripe/streams.py

@@ -7,6 +7,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Iterable, Mapping, MutableMapping, Optional
 
+import pendulum
 import requests
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams.http import HttpStream
@@ -57,6 +58,10 @@ class IncrementalStripeStream(StripeStream, ABC):
     # Stripe returns most recently created objects first, so we don't want to persist state until the entire stream has been read
     state_checkpoint_interval = math.inf
 
+    def __init__(self, start_date: str, **kwargs):
+        super().__init__(**kwargs)
+        self.start_date = pendulum.parse(start_date).int_timestamp
+
     @property
     @abstractmethod
     def cursor_field(self) -> str:
@@ -76,8 +81,14 @@ def get_updated_state(self, current_stream_state: MutableMapping[str, Any], late
     def request_params(self, stream_state: Mapping[str, Any] = None, **kwargs):
         stream_state = stream_state or {}
         params = super().request_params(stream_state=stream_state, **kwargs)
+
+        start_point = self.start_date
         if stream_state and self.cursor_field in stream_state:
-            params["created[gte]"] = stream_state.get(self.cursor_field)
+            start_point = max(start_point, stream_state[self.cursor_field])
+
+        if start_point:
+            params["created[gte]"] = start_point
+
         return params
 
 

docs/integrations/sources/stripe.md

@@ -12,7 +12,7 @@ This Source is capable of syncing the following core Streams:
 * [Bank accounts](https://stripe.com/docs/api/customer_bank_accounts/list)
 * [Charges](https://stripe.com/docs/api/charges/list) \(Incremental\)
 * [Coupons](https://stripe.com/docs/api/coupons/list) \(Incremental\)
-* [Customer Balance Transactions](https://stripe.com/docs/api/customer_balance_transactions/list) \(Incremental\)
+* [Customer Balance Transactions](https://stripe.com/docs/api/customer_balance_transactions/list)
 * [Customers](https://stripe.com/docs/api/customers/list) \(Incremental\)
 * [Disputes](https://stripe.com/docs/api/disputes/list) \(Incremental\)
 * [Events](https://stripe.com/docs/api/events/list) \(Incremental\)
@@ -72,11 +72,12 @@ If you would like to test Airbyte using test data on Stripe, `sk_test_` and `rk_
 
 | Version | Date       | Pull Request | Subject |
 | :------ | :--------  | :-----       | :------ |
-| 0.1.18   | 2021-09-14 | [6004](https://github.com/airbytehq/airbyte/pull/6004) | Fix  coupons and subscriptions stream schemas by removing incorrect timestamp formatting |
+| 0.1.19   | 2021-09-27 | [6466](https://github.com/airbytehq/airbyte/pull/6466) | Use `start_date` parameter in incremental streams |
+| 0.1.18   | 2021-09-14 | [6004](https://github.com/airbytehq/airbyte/pull/6004) | Fix coupons and subscriptions stream schemas by removing incorrect timestamp formatting |
 | 0.1.17   | 2021-09-14 | [6004](https://github.com/airbytehq/airbyte/pull/6004) | Add `PaymentIntents` stream |
 | 0.1.16   | 2021-07-28 | [4980](https://github.com/airbytehq/airbyte/pull/4980) | Remove Updated field from schemas |
-| 0.1.15   | 2021-07-21 | [4878](https://github.com/airbytehq/airbyte/pull/4878) | Fix incorrect percent_off and discounts data filed types|
-| 0.1.14   | 2021-07-09 | [4669](https://github.com/airbytehq/airbyte/pull/4669) | Subscriptions Stream now returns all kinds of subscriptions (including expired and canceled)|
+| 0.1.15   | 2021-07-21 | [4878](https://github.com/airbytehq/airbyte/pull/4878) | Fix incorrect percent_off and discounts data filed types |
+| 0.1.14   | 2021-07-09 | [4669](https://github.com/airbytehq/airbyte/pull/4669) | Subscriptions Stream now returns all kinds of subscriptions (including expired and canceled) |
 | 0.1.13   | 2021-07-03 | [4528](https://github.com/airbytehq/airbyte/pull/4528) | Remove regex for acc validation |
 | 0.1.12   | 2021-06-08 | [3973](https://github.com/airbytehq/airbyte/pull/3973) | Add `AIRBYTE_ENTRYPOINT` for Kubernetes support |
 | 0.1.11   | 2021-05-30 | [3744](https://github.com/airbytehq/airbyte/pull/3744) | Fix types in schema |