Chore: Update ruff rules and fix exceptions (#205)

Author: aaronsteers

airbyte/__init__.py

@@ -6,9 +6,19 @@
 ## API Reference
 
 """
+
 from __future__ import annotations
 
-from airbyte import caches, cloud, datasets, documents, exceptions, results, secrets, sources
+from airbyte import (
+    caches,
+    cloud,
+    datasets,
+    documents,
+    exceptions,  # noqa: ICN001  # No 'exc' alias for top-level module
+    results,
+    secrets,
+    sources,
+)
 from airbyte.caches.bigquery import BigQueryCache
 from airbyte.caches.duckdb import DuckDBCache
 from airbyte.caches.util import get_default_cache, new_local_cache

airbyte/_executor.py

@@ -124,7 +124,7 @@ def _stream_from_file(file: IO[str]) -> Generator[str, Any, None]:
         exit_code = process.wait()
 
         # If the exit code is not 0 or -15 (SIGTERM), raise an exception
-        if exit_code not in (0, -15):
+        if exit_code not in {0, -15}:
             raise exc.AirbyteSubprocessFailedError(
                 run_args=args,
                 exit_code=exit_code,

airbyte/_processors/__init__.py

@@ -0,0 +1,4 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Internal processors for PyAirbyte."""
+
+from __future__ import annotations

airbyte/_processors/base.py

@@ -185,7 +185,7 @@ def process_airbyte_messages(
 
             elif message.type is Type.STATE:
                 state_msg = cast(AirbyteStateMessage, message.state)
-                if state_msg.type in [AirbyteStateType.GLOBAL, AirbyteStateType.LEGACY]:
+                if state_msg.type in {AirbyteStateType.GLOBAL, AirbyteStateType.LEGACY}:
                     self._pending_state_messages[f"_{state_msg.type}"].append(state_msg)
                 else:
                     stream_state = cast(AirbyteStreamState, state_msg.stream)

airbyte/_processors/sql/base.py

@@ -104,7 +104,7 @@ def __init__(
         self._ensure_schema_exists()
         self._catalog_manager = CatalogManager(
             engine=self.get_sql_engine(),
-            table_name_resolver=lambda stream_name: self.get_sql_table_name(stream_name),
+            table_name_resolver=self.get_sql_table_name,
         )
         self.file_writer = file_writer or self.file_writer_class(cache)
         self.type_converter = self.type_converter_class()
@@ -550,7 +550,7 @@ def write_stream_data(
             for batch_handle in batches_to_finalize:
                 files += batch_handle.files
             # Use the max batch ID as the batch ID for table names.
-            max_batch_id = max([batch.batch_id for batch in batches_to_finalize])
+            max_batch_id = max(batch.batch_id for batch in batches_to_finalize)
 
             temp_table_name = self._write_files_to_new_table(
                 files=files,
@@ -820,7 +820,7 @@ def _get_primary_keys(
         joined_pks = [".".join(pk) for pk in pks]
         for pk in joined_pks:
             if "." in pk:
-                msg = "Nested primary keys are not yet supported. Found: {pk}"
+                msg = f"Nested primary keys are not yet supported. Found: {pk}"
                 raise NotImplementedError(msg)
 
         return joined_pks

airbyte/_processors/sql/duckdb.py

@@ -46,29 +46,6 @@ def _setup(self) -> None:
 
         Path(self.cache.db_path).parent.mkdir(parents=True, exist_ok=True)
 
-    @overrides
-    def _ensure_compatible_table_schema(
-        self,
-        stream_name: str,
-        *,
-        raise_on_error: bool = True,
-    ) -> bool:
-        """Return true if the given table is compatible with the stream's schema.
-
-        In addition to the base implementation, this also checks primary keys.
-        """
-        # call super
-        if not super()._ensure_compatible_table_schema(
-            stream_name=stream_name,
-            raise_on_error=raise_on_error,
-        ):
-            return False
-
-        # TODO: Add validation for primary keys after DuckDB adds support for primary key
-        #       inspection: https://github.com/Mause/duckdb_engine/issues/594
-
-        return True
-
     def _write_files_to_new_table(
         self,
         files: list[Path],

airbyte/_util/__init__.py

@@ -4,6 +4,7 @@
 Note: This module is for internal use only and it should not be depended upon for production use.
 It is subject to change without notice.
 """
+
 from __future__ import annotations
 
 from airbyte._util.pip_util import connector_pip_url, github_pip_url

airbyte/_util/document_rendering.py

@@ -1,5 +1,6 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 """Methods for converting Airbyte records into documents."""
+
 from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any, Optional

airbyte/_util/meta.py

@@ -3,6 +3,7 @@
 
 This module contains functions for detecting environment and runtime information.
 """
+
 from __future__ import annotations
 
 import os
@@ -61,10 +62,7 @@ def is_interactive() -> bool:
     if is_ci():
         return False
 
-    if sys.__stdin__.isatty() and sys.__stdout__.isatty():
-        return True
-
-    return False
+    return bool(sys.__stdin__.isatty() and sys.__stdout__.isatty())
 
 
 @lru_cache
@@ -105,7 +103,7 @@ def get_notebook_name() -> str | None:
 @lru_cache
 def get_vscode_notebook_name() -> str | None:
     with suppress(Exception):
-        import IPython
+        import IPython  # noqa: PLC0415
 
         return Path(
             IPython.extract_module_locals()[1]["__vsc_ipynb_file__"],

airbyte/_util/telemetry.py

@@ -28,6 +28,7 @@
 - Flags to help us understand if PyAirbyte is running on CI, Google Colab, or another environment.
 
 """
+
 from __future__ import annotations
 
 import datetime