chore: misc mypy typing fixes

Author: aaronsteers

.github/dependabot.yml

@@ -13,10 +13,10 @@ updates:
       interval: daily
     labels:
       - chore
-    open-pull-requests-limit: 8  # default is 5
+    open-pull-requests-limit: 8 # default is 5
 
   - package-ecosystem: github-actions
-    open-pull-requests-limit: 5  # default is 5
+    open-pull-requests-limit: 5 # default is 5
     directory: "/"
     commit-message:
       prefix: "ci(deps): "
@@ -29,5 +29,5 @@ updates:
       minor-and-patch:
         applies-to: version-updates
         update-types:
-        - patch
-        - minor
+          - patch
+          - minor

.github/workflows/publish_sdm_connector.yml

@@ -7,12 +7,11 @@ on:
   workflow_dispatch:
     inputs:
       version:
-        description:
-            The version to publish, ie 1.0.0 or 1.0.0-dev1.
-            If omitted, and if run from a release branch, the version will be
-            inferred from the git tag.
-            If omitted, and if run from a non-release branch, then only a SHA-based
-            Docker tag will be created.
+        description: The version to publish, ie 1.0.0 or 1.0.0-dev1.
+          If omitted, and if run from a release branch, the version will be
+          inferred from the git tag.
+          If omitted, and if run from a non-release branch, then only a SHA-based
+          Docker tag will be created.
         required: false
       dry_run:
         description: If true, the workflow will not push to DockerHub.
@@ -24,7 +23,6 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-
       - name: Detect Release Tag Version
         if: startsWith(github.ref, 'refs/tags/v')
         run: |
@@ -167,7 +165,6 @@ jobs:
           tags: |
             airbyte/source-declarative-manifest:${{ env.VERSION }}
 
-
       - name: Build and push ('latest' tag)
         # Only run if version is set and IS_PRERELEASE is false
         if: env.VERSION != '' && env.IS_PRERELEASE == 'false' && github.event.inputs.dry_run == 'false'

.github/workflows/pytest_matrix.yml

@@ -12,10 +12,10 @@ on:
     branches:
       - main
     paths:
-      - 'airbyte_cdk/**'
-      - 'unit_tests/**'
-      - 'poetry.lock'
-      - 'pyproject.toml'
+      - "airbyte_cdk/**"
+      - "unit_tests/**"
+      - "poetry.lock"
+      - "pyproject.toml"
   pull_request:
 
 jobs:

airbyte_cdk/cli/source_declarative_manifest/_run.py

@@ -160,10 +160,14 @@ def create_declarative_source(args: list[str]) -> ConcurrentDeclarativeSource:
     connector builder.
     """
     try:
+        config: Mapping[str, Any] | None
+        catalog: ConfiguredAirbyteCatalog | None
+        state: list[AirbyteStateMessage]
         config, catalog, state = _parse_inputs_into_config_catalog_state(args)
-        if "__injected_declarative_manifest" not in config:
+        if config is None or "__injected_declarative_manifest" not in config:
             raise ValueError(
-                f"Invalid config: `__injected_declarative_manifest` should be provided at the root of the config but config only has keys {list(config.keys())}"
+                "Invalid config: `__injected_declarative_manifest` should be provided at the root "
+                f"of the config but config only has keys: {list(config.keys() if config else [])}"
             )
         return ConcurrentDeclarativeSource(
             config=config,

airbyte_cdk/destinations/vector_db_based/writer.py

@@ -85,6 +85,9 @@ def write(
                 record_chunks, record_id_to_delete = self.processor.process(message.record)
                 self.chunks[(message.record.namespace, message.record.stream)].extend(record_chunks)
                 if record_id_to_delete is not None:
+                    if message.record is None:
+                        raise ValueError("Record messages cannot have null `record` property.")
+
                     self.ids_to_delete[(message.record.namespace, message.record.stream)].append(
                         record_id_to_delete
                     )

airbyte_cdk/sources/connector_state_manager.py

@@ -131,7 +131,7 @@ def _extract_from_state_message(
         else:
             streams = {
                 HashableStreamDescriptor(
-                    name=per_stream_state.stream.stream_descriptor.name,
+                    name=per_stream_state.stream.stream_descriptor.name,  # type: ignore[union-attr] # stream has stream_descriptor
                     namespace=per_stream_state.stream.stream_descriptor.namespace,  # type: ignore[union-attr] # stream has stream_descriptor
                 ): per_stream_state.stream.stream_state  # type: ignore[union-attr] # stream has stream_state
                 for per_stream_state in state

airbyte_cdk/sources/declarative/auth/selective_authenticator.py

@@ -28,7 +28,12 @@ def __new__(  # type: ignore[misc]
         **kwargs: Any,
     ) -> DeclarativeAuthenticator:
         try:
-            selected_key = str(dpath.get(config, authenticator_selection_path))
+            selected_key = str(
+                dpath.get(
+                    config,  # type: ignore [arg-type]  # Dpath wants mutable mapping but doesn't need it.
+                    authenticator_selection_path,
+                )
+            )
         except KeyError as err:
             raise ValueError(
                 "The path from `authenticator_selection_path` is not found in the config."

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2044,17 +2044,18 @@ definitions:
               The DeclarativeOAuth Specific URL templated string to obtain the `access_token`, `refresh_token` etc.
               The placeholders are replaced during the processing to provide neccessary values.
             examples:
-              - access_token_url: https://auth.host.com/oauth2/token?{client_id_key}={{client_id_key}}&{client_secret_key}={{client_secret_key}}&{auth_code_key}={{auth_code_key}}&{redirect_uri_key}={urlEncoder:{{redirect_uri_key}}} 
+              - access_token_url: https://auth.host.com/oauth2/token?{client_id_key}={{client_id_key}}&{client_secret_key}={{client_secret_key}}&{auth_code_key}={{auth_code_key}}&{redirect_uri_key}={urlEncoder:{{redirect_uri_key}}}
           access_token_headers:
             title: (Optional) DeclarativeOAuth Access Token Headers
             type: object
             additionalProperties: true
             description: |-
               The DeclarativeOAuth Specific optional headers to inject while exchanging the `auth_code` to `access_token` during `completeOAuthFlow` step.
             examples:
-              - access_token_headers: {
-                  "Authorization": "Basic {base64Encoder:{client_id}:{client_secret}}"
-                }
+              - access_token_headers:
+                  {
+                    "Authorization": "Basic {base64Encoder:{client_id}:{client_secret}}",
+                  }
           access_token_params:
             title: (Optional) DeclarativeOAuth Access Token Query Params (Json Encoded)
             type: object
@@ -2063,18 +2064,19 @@ definitions:
               The DeclarativeOAuth Specific optional query parameters to inject while exchanging the `auth_code` to `access_token` during `completeOAuthFlow` step.
               When this property is provided, the query params will be encoded as `Json` and included in the outgoing API request.
             examples:
-              - access_token_params: {
-                  "{auth_code_key}": "{{auth_code_key}}",
-                  "{client_id_key}": "{{client_id_key}}",
-                  "{client_secret_key}": "{{client_secret_key}}"
-                }
+              - access_token_params:
+                  {
+                    "{auth_code_key}": "{{auth_code_key}}",
+                    "{client_id_key}": "{{client_id_key}}",
+                    "{client_secret_key}": "{{client_secret_key}}",
+                  }
           extract_output:
             title: DeclarativeOAuth Extract Output
             type: array
             items:
               type: string
             description: |-
-              The DeclarativeOAuth Specific list of strings to indicate which keys should be extracted and returned back to the input config.                
+              The DeclarativeOAuth Specific list of strings to indicate which keys should be extracted and returned back to the input config.
             examples:
               - extract_output: ["access_token", "refresh_token", "other_field"]
           state:
@@ -2086,17 +2088,14 @@ definitions:
               - max
             description: |-
               The DeclarativeOAuth Specific object to provide the criteria of how the `state` query param should be constructed,
-              including length and complexity.                
+              including length and complexity.
             properties:
               min:
                 type: integer
               max:
                 type: integer
             examples:
-              - state: {
-                  "min": 7,
-                  "max": 128,
-                }
+              - state: { "min": 7, "max": 128 }
           client_id_key:
             title: (Optional) DeclarativeOAuth Client ID Key Override
             type: string
@@ -2122,14 +2121,14 @@ definitions:
             title: (Optional) DeclarativeOAuth State Key Override
             type: string
             description: |-
-              The DeclarativeOAuth Specific optional override to provide the custom `state` key name, if required by data-provider.                
+              The DeclarativeOAuth Specific optional override to provide the custom `state` key name, if required by data-provider.
             examples:
               - state_key: "my_custom_state_key_key_name"
           auth_code_key:
             title: (Optional) DeclarativeOAuth Auth Code Key Override
             type: string
             description: |-
-              The DeclarativeOAuth Specific optional override to provide the custom `code` key name to something like `auth_code` or `custom_auth_code`, if required by data-provider.                
+              The DeclarativeOAuth Specific optional override to provide the custom `code` key name to something like `auth_code` or `custom_auth_code`, if required by data-provider.
             examples:
               - auth_code_key: "my_custom_auth_code_key_name"
           redirect_uri_key:

airbyte_cdk/sources/declarative/decoders/noop_decoder.py

@@ -14,5 +14,8 @@ class NoopDecoder(Decoder):
     def is_stream_response(self) -> bool:
         return False
 
-    def decode(self, response: requests.Response) -> Generator[Mapping[str, Any], None, None]:
+    def decode(  # type: ignore[override]
+        self,
+        response: requests.Response,
+    ) -> Generator[Mapping[str, Any], None, None]:
         yield from [{}]

airbyte_cdk/sources/file_based/availability_strategy/abstract_file_based_availability_strategy.py

@@ -22,8 +22,11 @@
 
 class AbstractFileBasedAvailabilityStrategy(AvailabilityStrategy):
     @abstractmethod
-    def check_availability(
-        self, stream: Stream, logger: logging.Logger, _: Optional[Source]
+    def check_availability(  # type: ignore[override]
+        self,
+        stream: Stream,
+        logger: logging.Logger,
+        _: Optional[Source],
     ) -> Tuple[bool, Optional[str]]:
         """
         Perform a connection check for the stream.
@@ -34,7 +37,10 @@ def check_availability(
 
     @abstractmethod
     def check_availability_and_parsability(
-        self, stream: "AbstractFileBasedStream", logger: logging.Logger, _: Optional[Source]
+        self,
+        stream: AbstractFileBasedStream,
+        logger: logging.Logger,
+        _: Optional[Source],
     ) -> Tuple[bool, Optional[str]]:
         """
         Performs a connection check for the stream, as well as additional checks that
@@ -46,7 +52,7 @@ def check_availability_and_parsability(
 
 
 class AbstractFileBasedAvailabilityStrategyWrapper(AbstractAvailabilityStrategy):
-    def __init__(self, stream: "AbstractFileBasedStream"):
+    def __init__(self, stream: AbstractFileBasedStream) -> None:
         self.stream = stream
 
     def check_availability(self, logger: logging.Logger) -> StreamAvailability:

airbyte_cdk/sources/file_based/availability_strategy/default_file_based_availability_strategy.py

@@ -28,9 +28,12 @@ class DefaultFileBasedAvailabilityStrategy(AbstractFileBasedAvailabilityStrategy
     def __init__(self, stream_reader: AbstractFileBasedStreamReader):
         self.stream_reader = stream_reader
 
-    def check_availability(
-        self, stream: "AbstractFileBasedStream", logger: logging.Logger, _: Optional[Source]
-    ) -> Tuple[bool, Optional[str]]:  # type: ignore[override]
+    def check_availability(  # type: ignore[override]
+        self,
+        stream: AbstractFileBasedStream,
+        logger: logging.Logger,
+        _: Optional[Source],
+    ) -> Tuple[bool, Optional[str]]:
         """
         Perform a connection check for the stream (verify that we can list files from the stream).
 
@@ -44,7 +47,10 @@ def check_availability(
         return True, None
 
     def check_availability_and_parsability(
-        self, stream: "AbstractFileBasedStream", logger: logging.Logger, _: Optional[Source]
+        self,
+        stream: AbstractFileBasedStream,
+        logger: logging.Logger,
+        _: Optional[Source],
     ) -> Tuple[bool, Optional[str]]:
         """
         Perform a connection check for the stream.
@@ -82,7 +88,7 @@ def check_availability_and_parsability(
 
         return True, None
 
-    def _check_list_files(self, stream: "AbstractFileBasedStream") -> RemoteFile:
+    def _check_list_files(self, stream: AbstractFileBasedStream) -> RemoteFile:
         """
         Check that we can list files from the stream.
 
@@ -102,7 +108,10 @@ def _check_list_files(self, stream: "AbstractFileBasedStream") -> RemoteFile:
         return file
 
     def _check_parse_record(
-        self, stream: "AbstractFileBasedStream", file: RemoteFile, logger: logging.Logger
+        self,
+        stream: AbstractFileBasedStream,
+        file: RemoteFile,
+        logger: logging.Logger,
     ) -> None:
         parser = stream.get_parser()
 

airbyte_cdk/sources/file_based/file_types/avro_parser.py

@@ -3,7 +3,7 @@
 #
 
 import logging
-from typing import Any, Dict, Iterable, Mapping, Optional, Tuple
+from typing import Any, Dict, Iterable, Mapping, Optional, Tuple, cast
 
 import fastavro
 
@@ -183,7 +183,7 @@ def parse_records(
                 avro_reader = fastavro.reader(fp)
                 schema = avro_reader.writer_schema
                 schema_field_name_to_type = {
-                    field["name"]: field["type"] for field in schema["fields"]
+                    field["name"]: cast(dict, field["type"]) for field in schema["fields"]
                 }
                 for record in avro_reader:
                     line_no += 1

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -218,7 +218,9 @@ def observe(self, record: Record) -> None:
         )
         cursor_value = self._extract_cursor_value(record)
 
-        if most_recent_cursor_value is None or most_recent_cursor_value < cursor_value:
+        if record.associated_slice is not None and (
+            most_recent_cursor_value is None or most_recent_cursor_value < cursor_value
+        ):
             self._most_recent_cursor_value_per_partition[record.associated_slice] = cursor_value
 
     def _extract_cursor_value(self, record: Record) -> Any:

airbyte_cdk/sources/streams/core.py

@@ -115,11 +115,6 @@ class StreamClassification:
     has_multiple_slices: bool
 
 
-# Moved to class declaration since get_updated_state is called on every record for incremental syncs, and thus the @deprecated decorator as well.
-@deprecated(
-    "Deprecated as of CDK version 0.1.49. ",
-    "Deprecated method get_updated_state, You should use explicit state property instead, see IncrementalMixin docs.",
-)
 class Stream(ABC):
     """
     Base abstract class for an Airbyte Stream. Makes no assumption of the Stream's underlying transport protocol.
@@ -435,6 +430,10 @@ def state_checkpoint_interval(self) -> Optional[int]:
         """
         return None
 
+    @deprecated(
+        "Deprecated method `get_updated_state` as of CDK version 0.1.49. "
+        "Please use explicit state property instead, see `IncrementalMixin` docs.",
+    )
     def get_updated_state(
         self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]
     ) -> MutableMapping[str, Any]:

airbyte_cdk/sources/streams/http/requests_native_auth/abstract_token.py

@@ -5,13 +5,14 @@
 from abc import abstractmethod
 from typing import Any, Mapping
 
+import requests
 from requests.auth import AuthBase
 
 
 class AbstractHeaderAuthenticator(AuthBase):
     """Abstract class for an header-based authenticators that add a header to outgoing HTTP requests."""
 
-    def __call__(self, request):
+    def __call__(self, request: requests.PreparedRequest) -> Any:
         """Attach the HTTP headers required to authenticate on the HTTP request"""
         request.headers.update(self.get_auth_header())
         return request