refactor(source-mixpanel): migrate to CDK v3 (#41969)

Author: lazebnyi

airbyte-integrations/connectors/source-mixpanel/metadata.yaml

@@ -11,7 +11,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 12928b32-bf0a-4f1e-964f-07e12e37153a
-  dockerImageTag: 3.3.3
+  dockerImageTag: 3.4.0
   dockerRepository: airbyte/source-mixpanel
   documentationUrl: https://docs.airbyte.com/integrations/sources/mixpanel
   githubIssueLabel: source-mixpanel

airbyte-integrations/connectors/source-mixpanel/poetry.lock

@@ -3,7 +3,7 @@ requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "3.3.3"
+version = "3.4.0"
 name = "source-mixpanel"
 description = "Source implementation for Mixpanel."
 authors = ["Airbyte <[email protected]>"]
@@ -16,8 +16,8 @@ repository = "https://github.com/airbytehq/airbyte"
 include = "source_mixpanel"
 
 [tool.poetry.dependencies]
-python = "^3.9,<3.12"
-airbyte-cdk = "0.80.0"
+python = "^3.10,<3.12"
+airbyte-cdk = "^4"
 
 [tool.poetry.scripts]
 source-mixpanel = "source_mixpanel.run:run"

airbyte-integrations/connectors/source-mixpanel/pyproject.toml

@@ -0,0 +1,25 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from typing import Any, Optional, Union
+
+import requests
+from airbyte_cdk import BackoffStrategy
+from airbyte_cdk.sources.streams.http import HttpStream
+
+
+class MixpanelStreamBackoffStrategy(BackoffStrategy):
+    def __init__(self, stream: HttpStream, **kwargs):  # type: ignore # noqa
+        self.stream = stream
+        super().__init__(**kwargs)
+
+    def backoff_time(
+        self, response_or_exception: Optional[Union[requests.Response, requests.RequestException]], **kwargs: Any
+    ) -> Optional[float]:
+        if isinstance(response_or_exception, requests.Response):
+            retry_after = response_or_exception.headers.get("Retry-After")
+            if retry_after:
+                self._logger.debug(f"API responded with `Retry-After` header: {retry_after}")
+                return float(retry_after)
+        return None

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/backoff_strategy.py

@@ -277,7 +277,7 @@ class EngagePaginationStrategy(PageIncrement):
 
     _total = 0
 
-    def next_page_token(self, response, last_records: List[Mapping[str, Any]]) -> Optional[Mapping[str, Any]]:
+    def next_page_token(self, response: requests.Response, last_page_size: int, last_record: Optional[Record]) -> Optional[Any]:
         """
         Determines page and subpage numbers for the `items` stream
 

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/components.py

@@ -0,0 +1,7 @@
+from .base_errors_handler import MixpanelStreamErrorHandler
+from .export_errors_handler import ExportErrorHandler
+
+__all__ = [
+    "MixpanelStreamErrorHandler",
+    "ExportErrorHandler",
+]

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/errors_handlers/__init__.py

@@ -0,0 +1,44 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+import re
+from typing import Optional, Union
+
+import requests
+from airbyte_cdk.sources.streams.http import HttpStream
+from airbyte_cdk.sources.streams.http.error_handlers import ErrorResolution, HttpStatusErrorHandler, ResponseAction
+from airbyte_protocol.models import FailureType
+
+
+class MixpanelStreamErrorHandler(HttpStatusErrorHandler):
+    """
+    Custom error handler for Mixpanel stream that interprets responses and exceptions.
+
+    This handler specifically addresses:
+    - 400 status code with "Unable to authenticate request" message, indicating potential credential expiration.
+    - 402 status code, indicating a payment required error.
+
+    If the response does not match these specific cases, the handler defers to the parent class's implementation.
+    """
+
+    def interpret_response(self, response_or_exception: Optional[Union[requests.Response, Exception]] = None) -> ErrorResolution:
+        if isinstance(response_or_exception, requests.Response):
+            if response_or_exception.status_code == 400 and "Unable to authenticate request" in response_or_exception.text:
+                message = (
+                    f"Your credentials might have expired. Please update your config with valid credentials."
+                    f" See more details: {response_or_exception.text}"
+                )
+                return ErrorResolution(
+                    response_action=ResponseAction.FAIL,
+                    failure_type=FailureType.config_error,
+                    error_message=message,
+                )
+            elif response_or_exception.status_code == 402:
+                message = f"Unable to perform a request. Payment Required: {response_or_exception.json()['error']}"
+                return ErrorResolution(
+                    response_action=ResponseAction.FAIL,
+                    failure_type=FailureType.transient_error,
+                    error_message=message,
+                )
+        return super().interpret_response(response_or_exception)

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/errors_handlers/base_errors_handler.py

@@ -0,0 +1,55 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from typing import Optional, Union
+
+import requests
+from airbyte_cdk.sources.streams.http import HttpStream
+from airbyte_cdk.sources.streams.http.error_handlers import ErrorResolution, HttpStatusErrorHandler, ResponseAction
+from airbyte_protocol.models import FailureType
+
+from .base_errors_handler import MixpanelStreamErrorHandler
+
+
+class ExportErrorHandler(MixpanelStreamErrorHandler):
+    """
+    Custom error handler for handling export errors specific to Mixpanel streams.
+
+    This handler addresses:
+    - 400 status code with "to_date cannot be later than today" message, indicating a potential timezone mismatch.
+    - ConnectionResetError during response parsing, indicating a need to retry the request.
+
+    If the response does not match these specific cases, the handler defers to the parent class's implementation.
+
+    Attributes:
+        stream (HttpStream): The HTTP stream associated with this error handler.
+    """
+
+    def __init__(self, stream: HttpStream, **kwargs):  # type: ignore # noqa
+        self.stream = stream
+        super().__init__(**kwargs)
+
+    def interpret_response(self, response_or_exception: Optional[Union[requests.Response, Exception]] = None) -> ErrorResolution:
+        if isinstance(response_or_exception, requests.Response):
+            if (
+                response_or_exception.status_code == requests.codes.bad_request
+                and "to_date cannot be later than today" in response_or_exception.text
+            ):
+                self.stream._timezone_mismatch = True
+                message = "Your project timezone must be misconfigured. Please set it to the one defined in your Mixpanel project settings. Stopping current stream sync."
+                return ErrorResolution(
+                    response_action=ResponseAction.IGNORE,
+                    failure_type=FailureType.config_error,
+                    error_message=message,
+                )
+            try:
+                # trying to parse response to avoid ConnectionResetError and retry if it occurs
+                self.stream.iter_dicts(response_or_exception.iter_lines(decode_unicode=True))
+            except ConnectionResetError:
+                return ErrorResolution(
+                    response_action=ResponseAction.RETRY,
+                    failure_type=FailureType.transient_error,
+                    error_message=f"Response status code: {response_or_exception.status_code}. Retrying...",
+                )
+        return super().interpret_response(response_or_exception)

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/errors_handlers/export_errors_handler.py

@@ -39,10 +39,10 @@ definitions:
         action: FAIL
         error_message: Unable to perform a request. Payment Required.
       - predicate: "{{ 'Retry-After' in headers }}"
-        action: RETRY
+        action: RATE_LIMITED
         error_message: Query rate limit exceeded.
       - error_message_contains: "Query rate limit exceeded"
-        action: RETRY
+        action: RATE_LIMITED
         error_message: Query rate limit exceeded.
       - http_codes: [500]
         error_message_contains: "unknown error"

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/manifest.yaml

@@ -9,11 +9,13 @@
 
 import pendulum
 import requests
-from airbyte_cdk.models import FailureType
+from airbyte_cdk import BackoffStrategy
 from airbyte_cdk.sources.streams.http import HttpStream
-from airbyte_cdk.utils import AirbyteTracedException
+from airbyte_cdk.sources.streams.http.error_handlers import ErrorHandler
 from pendulum import Date
 from requests.auth import AuthBase
+from source_mixpanel.backoff_strategy import MixpanelStreamBackoffStrategy
+from source_mixpanel.errors_handlers import MixpanelStreamErrorHandler
 from source_mixpanel.utils import fix_date_time
 
 
@@ -69,7 +71,6 @@ def __init__(
         self.region = region
         self.project_timezone = project_timezone
         self.project_id = project_id
-        self.retries = 0
         self._reqs_per_hour_limit = reqs_per_hour_limit
         super().__init__(authenticator=authenticator)
 
@@ -110,42 +111,18 @@ def parse_response(
             self.logger.info(f"Sleep for {3600 / self.reqs_per_hour_limit} seconds to match API limitations after reading from {self.name}")
             time.sleep(3600 / self.reqs_per_hour_limit)
 
-    @property
-    def max_retries(self) -> Union[int, None]:
-        # we want to limit the max sleeping time by 2^3 * 60 = 8 minutes
-        return 3
+    def get_backoff_strategy(self) -> Optional[Union[BackoffStrategy, List[BackoffStrategy]]]:
+        return MixpanelStreamBackoffStrategy(stream=self)
 
-    def backoff_time(self, response: requests.Response) -> float:
-        """
-        Some API endpoints do not return "Retry-After" header.
-        """
-
-        retry_after = response.headers.get("Retry-After")
-        if retry_after:
-            self.logger.debug(f"API responded with `Retry-After` header: {retry_after}")
-            return float(retry_after)
-
-        self.retries += 1
-        return 2**self.retries * 60
-
-    def should_retry(self, response: requests.Response) -> bool:
-        if response.status_code == 402:
-            self.logger.warning(f"Unable to perform a request. Payment Required: {response.json()['error']}")
-            return False
-        if response.status_code == 400 and "Unable to authenticate request" in response.text:
-            message = (
-                f"Your credentials might have expired. Please update your config with valid credentials."
-                f" See more details: {response.text}"
-            )
-            raise AirbyteTracedException(message=message, internal_message=message, failure_type=FailureType.config_error)
-        return super().should_retry(response)
+    def get_error_handler(self) -> Optional[ErrorHandler]:
+        return MixpanelStreamErrorHandler(logger=self.logger)
 
     def get_stream_params(self) -> Mapping[str, Any]:
         """
         Fetch required parameters in a given stream. Used to create sub-streams
         """
         params = {
-            "authenticator": self._session.auth,
+            "authenticator": self._http_client._session.auth,
             "region": self.region,
             "project_timezone": self.project_timezone,
             "reqs_per_hour_limit": self.reqs_per_hour_limit,
@@ -168,18 +145,6 @@ def request_params(
 class DateSlicesMixin:
     raise_on_http_errors = True
 
-    def should_retry(self, response: requests.Response) -> bool:
-        if response.status_code == requests.codes.bad_request:
-            if "to_date cannot be later than today" in response.text:
-                self._timezone_mismatch = True
-                self.logger.warning(
-                    "Your project timezone must be misconfigured. Please set it to the one defined in your Mixpanel project settings. "
-                    "Stopping current stream sync."
-                )
-                setattr(self, "raise_on_http_errors", False)
-                return False
-        return super().should_retry(response)
-
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._timezone_mismatch = False

airbyte-integrations/connectors/source-mixpanel/source_mixpanel/streams/base.py

@@ -4,14 +4,16 @@
 
 import json
 from functools import cache
-from typing import Any, Iterable, Mapping, MutableMapping
+from typing import Any, Iterable, Mapping, MutableMapping, Optional
 
 import pendulum
 import requests
 from airbyte_cdk.models import SyncMode
+from airbyte_cdk.sources.streams.http.error_handlers import ErrorHandler
 from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
+from source_mixpanel.errors_handlers import ExportErrorHandler
+from source_mixpanel.property_transformation import transform_property_names
 
-from ..property_transformation import transform_property_names
 from .base import DateSlicesMixin, IncrementalMixpanelStream, MixpanelStream
 
 
@@ -88,13 +90,8 @@ def url_base(self):
     def path(self, **kwargs) -> str:
         return "export"
 
-    def should_retry(self, response: requests.Response) -> bool:
-        try:
-            # trying to parse response to avoid ConnectionResetError and retry if it occurs
-            self.iter_dicts(response.iter_lines(decode_unicode=True))
-        except ConnectionResetError:
-            return True
-        return super().should_retry(response)
+    def get_error_handler(self) -> Optional[ErrorHandler]:
+        return ExportErrorHandler(logger=self.logger, stream=self)
 
     def iter_dicts(self, lines):
         """