source-stripe: pass type checks

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

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: e094cb9a-26de-4645-8761-65c0c425d1de
-  dockerImageTag: 5.2.4
+  dockerImageTag: 5.2.5
   dockerRepository: airbyte/source-stripe
   documentationUrl: https://docs.airbyte.com/integrations/sources/stripe
   githubIssueLabel: source-stripe

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

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "5.2.4"
+version = "5.2.5"
 name = "source-stripe"
 description = "Source implementation for Stripe."
 authors = [ "Airbyte <contact@airbyte.io>",]
@@ -20,6 +20,7 @@ python = "^3.9,<3.12"
 stripe = "==2.56.0"
 pendulum = "==2.1.2"
 airbyte-cdk = "==0.60.1"
+mypy = "^1.8.0"
 
 [tool.poetry.scripts]
 source-stripe = "source_stripe.run:run"
@@ -29,3 +30,5 @@ requests-mock = "^1.11.0"
 pytest = "^6.1"
 freezegun = "==1.2.2"
 pytest-mock = "^3.6.1"
+types-stripe = "^3.5.2.20240106"
+types-requests = "^2.31.0.20240218"

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

@@ -8,6 +8,7 @@
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources import Source
 from airbyte_cdk.sources.streams import Stream
+from airbyte_cdk.sources.streams.http import HttpStream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 from requests import HTTPError
 
@@ -29,8 +30,9 @@ def _check_availability_for_sync_mode(
         sync_mode: SyncMode,
         logger: logging.Logger,
         source: Optional["Source"],
-        stream_state: Optional[Mapping[str, Any]],
+        stream_state: Optional[Mapping[Any, Any]],
     ) -> Tuple[bool, Optional[str]]:
+        reason: Optional[str] = None
         try:
             # Some streams need a stream slice to read records (e.g. if they have a SubstreamPartitionRouter)
             # Streams that don't need a stream slice will return `None` as their first stream slice.
@@ -90,7 +92,12 @@ def handle_http_error(
             raise error
         doc_ref = self._visit_docs_message(logger, source)
         reason = f"The endpoint {error.response.url} returned {status_code}: {error.response.reason}. {error_message}. {doc_ref} "
-        response_error_message = stream.parse_response_error_message(error.response)
+        # TODO alafanechere
+        # Can we make the HTTPAvailabilityStrategy handle HttpStream instead of Stream?
+        # The parse_response_error_message method is only available on HttpStream
+        response_error_message = None
+        if isinstance(stream, HttpStream):
+            response_error_message = stream.parse_response_error_message(error.response)
         if response_error_message:
             reason += response_error_message
         return False, reason

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

@@ -6,14 +6,14 @@
 import sys
 import traceback
 from datetime import datetime
-from typing import List
+from typing import List, Optional
 
 from airbyte_cdk.entrypoint import AirbyteEntrypoint, launch
 from airbyte_cdk.models import AirbyteErrorTraceMessage, AirbyteMessage, AirbyteTraceMessage, TraceType, Type
 from source_stripe import SourceStripe
 
 
-def _get_source(args: List[str]):
+def _get_source(args: List[str]) -> Optional[SourceStripe]:
     catalog_path = AirbyteEntrypoint.extract_catalog(args)
     config_path = AirbyteEntrypoint.extract_config(args)
     state_path = AirbyteEntrypoint.extract_state(args)
@@ -40,7 +40,7 @@ def _get_source(args: List[str]):
         return None
 
 
-def run():
+def run() -> None:
     _args = sys.argv[1:]
     source = _get_source(_args)
     if source:

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

@@ -5,11 +5,10 @@
 import logging
 import os
 from datetime import timedelta
-from typing import Any, List, Mapping, MutableMapping, Optional, Tuple
+from typing import Any, List, Mapping, MutableMapping, Optional, Tuple, Union
 
 import pendulum
 import stripe
-from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.entrypoint import logger as entrypoint_logger
 from airbyte_cdk.models import ConfiguredAirbyteCatalog, FailureType
 from airbyte_cdk.sources.concurrent_source.concurrent_source import ConcurrentSource
@@ -24,7 +23,7 @@
 from airbyte_cdk.sources.streams.concurrent.state_converters.datetime_stream_state_converter import EpochValueConcurrentStreamStateConverter
 from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
-from airbyte_protocol.models import SyncMode
+from airbyte_protocol.models import SyncMode  # type: ignore
 from source_stripe.streams import (
     CreatedCursorIncrementalStripeStream,
     CustomerBalanceTransactions,
@@ -61,7 +60,13 @@ class SourceStripe(ConcurrentSourceAdapter):
         CreatedCursorIncrementalStripeStream: ("created[gte]", "created[lte]"),
     }
 
-    def __init__(self, catalog: Optional[ConfiguredAirbyteCatalog], config: Optional[Mapping[str, Any]], state: TState, **kwargs):
+    def __init__(
+        self,
+        catalog: Optional[ConfiguredAirbyteCatalog],
+        config: Optional[Mapping[str, Any]],
+        state: Union[list[Any], MutableMapping[str, Any], None],
+        **kwargs: Any,
+    ):
         if config:
             concurrency_level = min(config.get("num_workers", _DEFAULT_CONCURRENCY), _MAX_CONCURRENCY)
         else:
@@ -83,13 +88,14 @@ def __init__(self, catalog: Optional[ConfiguredAirbyteCatalog], config: Optional
             self._streams_configured_as_full_refresh = set()
 
     @staticmethod
-    def validate_and_fill_with_defaults(config: MutableMapping[str, Any]) -> MutableMapping[str, Any]:
+    def validate_and_fill_with_defaults(config: Mapping[str, Any]) -> Mapping[str, Any]:
+        mutable_config = dict(config)
         lookback_window_days, slice_range = (
             config.get("lookback_window_days"),
             config.get("slice_range"),
         )
         if lookback_window_days is None:
-            config["lookback_window_days"] = 0
+            mutable_config["lookback_window_days"] = 0
         elif not isinstance(lookback_window_days, int) or lookback_window_days < 0:
             message = f"Invalid lookback window {lookback_window_days}. Please use only positive integer values or 0."
             raise AirbyteTracedException(
@@ -99,20 +105,20 @@ def validate_and_fill_with_defaults(config: MutableMapping[str, Any]) -> Mutable
             )
 
         # verifies the start_date in the config is valid
-        SourceStripe._start_date_to_timestamp(config)
+        SourceStripe._start_date_to_timestamp(mutable_config)
         if slice_range is None:
-            config["slice_range"] = 365
+            mutable_config["slice_range"] = 365
         elif not isinstance(slice_range, int) or slice_range < 1:
             message = f"Invalid slice range value {slice_range}. Please use positive integer values only."
             raise AirbyteTracedException(
                 message=message,
                 internal_message=message,
                 failure_type=FailureType.config_error,
             )
-        return config
+        return mutable_config
 
-    def check_connection(self, logger: AirbyteLogger, config: MutableMapping[str, Any]) -> Tuple[bool, Any]:
-        self.validate_and_fill_with_defaults(config)
+    def check_connection(self, logger: logging.Logger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
+        config = self.validate_and_fill_with_defaults(config)
         stripe.api_key = config["client_secret"]
         try:
             stripe.Account.retrieve(config["account_id"])
@@ -121,7 +127,7 @@ def check_connection(self, logger: AirbyteLogger, config: MutableMapping[str, An
         return True, None
 
     @staticmethod
-    def customers(**args):
+    def customers(**args: Any) -> IncrementalStripeStream:
         # The Customers stream is instantiated in a dedicated method to allow parametrization and avoid duplicated code.
         # It can be used with and without expanded items (as an independent stream or as a parent stream for other streams).
         return IncrementalStripeStream(
@@ -178,7 +184,7 @@ def get_api_call_budget(self, config: Mapping[str, Any]) -> AbstractAPIBudget:
 
         return HttpAPIBudget(policies=policies)
 
-    def streams(self, config: MutableMapping[str, Any]) -> List[Stream]:
+    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         config = self.validate_and_fill_with_defaults(config)
         authenticator = TokenAuthenticator(config["client_secret"])
 
@@ -522,7 +528,7 @@ def streams(self, config: MutableMapping[str, Any]) -> List[Stream]:
         state_manager = ConnectorStateManager(stream_instance_map={s.name: s for s in streams}, state=self._state)
         return [self._to_concurrent(stream, self._start_date_to_timestamp(config), state_manager) for stream in streams]
 
-    def _to_concurrent(self, stream: Stream, fallback_start, state_manager: ConnectorStateManager) -> Stream:
+    def _to_concurrent(self, stream: Stream, fallback_start: Any, state_manager: ConnectorStateManager) -> Stream:
         if stream.name in self._streams_configured_as_full_refresh:
             return StreamFacade.create_from_stream(stream, self, entrypoint_logger, self._create_empty_state(), NoopCursor())
 

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

@@ -8,7 +8,7 @@
 from airbyte_cdk.sources.streams.core import Stream, StreamData
 
 
-def get_first_stream_slice(stream, sync_mode, stream_state) -> Optional[Mapping[str, Any]]:
+def get_first_stream_slice(stream: Stream, sync_mode: SyncMode, stream_state: Optional[Mapping[str, Any]]) -> Optional[Mapping[str, Any]]:
     """
     Gets the first stream_slice from a given stream's stream_slices.
     :param stream: stream
@@ -17,9 +17,12 @@ def get_first_stream_slice(stream, sync_mode, stream_state) -> Optional[Mapping[
     :raises StopIteration: if there is no first slice to return (the stream_slices generator is empty)
     :return: first stream slice from 'stream_slices' generator (`None` is a valid stream slice)
     """
+    cursor_field: Optional[list[str]] = None
+    if isinstance(stream.cursor_field, str):
+        cursor_field = [stream.cursor_field]
     # We wrap the return output of stream_slices() because some implementations return types that are iterable,
     # but not iterators such as lists or tuples
-    slices = iter(stream.stream_slices(sync_mode=sync_mode, cursor_field=stream.cursor_field, stream_state=stream_state))
+    slices = iter(stream.stream_slices(sync_mode=sync_mode, cursor_field=cursor_field, stream_state=stream_state))
     return next(slices)