From 011ec41d0ce8a02689396f937b3e2738ca605120 Mon Sep 17 00:00:00 2001
From: Alexandros Milaios <alexmilaios@gmail.com>
Date: Fri, 4 Oct 2024 09:29:22 +0300
Subject: [PATCH] fix: add logic to slow down api calls to avoid rate limiting
 (#208)

---
 .../connectors/source-recurly/setup.py        |  2 +-
 .../source-recurly/source_recurly/streams.py  | 27 +++++++++++++------
 2 files changed, 20 insertions(+), 9 deletions(-)

diff --git a/airbyte-integrations/connectors/source-recurly/setup.py b/airbyte-integrations/connectors/source-recurly/setup.py
index d866d9361d1b..6e652147b0e7 100644
--- a/airbyte-integrations/connectors/source-recurly/setup.py
+++ b/airbyte-integrations/connectors/source-recurly/setup.py
@@ -5,7 +5,7 @@
 
 from setuptools import find_packages, setup
 
-MAIN_REQUIREMENTS = ["airbyte-cdk~=0.1", "recurly==4.10.0", "requests"]
+MAIN_REQUIREMENTS = ["airbyte-cdk==0.67", "recurly==4.10.0", "requests"]
 
 TEST_REQUIREMENTS = [
     "pytest~=6.1",
diff --git a/airbyte-integrations/connectors/source-recurly/source_recurly/streams.py b/airbyte-integrations/connectors/source-recurly/source_recurly/streams.py
index f7526476f4ae..63af3ee40dac 100644
--- a/airbyte-integrations/connectors/source-recurly/source_recurly/streams.py
+++ b/airbyte-integrations/connectors/source-recurly/source_recurly/streams.py
@@ -3,6 +3,7 @@
 #
 
 import re
+import time
 from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Union
 
 from airbyte_cdk.models import SyncMode
@@ -17,6 +18,7 @@
 
 BEGIN_TIME_PARAM = "begin_time"
 END_TIME_PARAM = "end_time"
+RATE_LIMIT_SLEEP_IN_SECS = 0.16
 
 CAMEL_CASE_PATTERN = re.compile(r"(?<!^)(?=[A-Z])")
 
@@ -128,11 +130,15 @@ def read_records(
         if self.end_time:
             params.update({END_TIME_PARAM: self.end_time})
 
-        items = getattr(self._client, self.client_method_name)(params=params).items()
+        pages = getattr(self._client, self.client_method_name)(params=params).pages()
 
         # Call the Recurly client methods
-        for item in items:
-            yield self._item_to_dict(item)
+        for page in pages:
+            for item in page:
+                yield self._item_to_dict(item)
+            # slow down the API calls to avoid rate limiting
+            # https://recurly.com/developers/api-v2/v2.2/#section/Rate-Limits
+            time.sleep(RATE_LIMIT_SLEEP_IN_SECS)
 
     def get_updated_state(self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]):
         """
@@ -205,9 +211,12 @@ def read_records(
         # and log a warn
         try:
             for account in accounts:
-                items = getattr(self._client, self.client_method_name)(params=params, account_id=account.id).items()
-                for item in items:
-                    yield self._item_to_dict(item)
+                pages = getattr(self._client, self.client_method_name)(params=params, account_id=account.id).pages()
+                for page in pages:
+                    for item in page:
+                        yield self._item_to_dict(item)
+                    # slow down the API calls to avoid rate limiting
+                    time.sleep(RATE_LIMIT_SLEEP_IN_SECS)
         except MissingFeatureError as error:
             super().logger.warning(f"Missing feature error {error}")
 
@@ -330,8 +339,10 @@ def read_records(
 
         for coupon in coupons:
             try:
-                items = self._client.list_unique_coupon_codes(params=params, coupon_id=coupon.id).items()
-                for item in items:
+                pages = self._client.list_unique_coupon_codes(params=params, coupon_id=coupon.id).pages()
+                for item in pages:
                     yield self._item_to_dict(item)
+                # slow down the API calls to avoid rate limiting
+                time.sleep(RATE_LIMIT_SLEEP_IN_SECS)
             except (NotFoundError, ValidationError):
                 pass