Chore: Configure warnings as errors in pytest (#245)

airbyte/_processors/sql/bigquery.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import warnings
 from pathlib import Path
 from typing import TYPE_CHECKING, Optional, final
 
@@ -56,7 +57,17 @@ def dataset_name(self) -> str:
 
     @overrides
     def get_sql_alchemy_url(self) -> SecretString:
-        """Return the SQLAlchemy URL to use."""
+        """Return the SQLAlchemy URL to use.
+
+        We suppress warnings about unrecognized JSON type. More info on that here:
+        - https://github.com/airbytehq/PyAirbyte/issues/254
+        """
+        warnings.filterwarnings(
+            "ignore",
+            message="Did not recognize type 'JSON' of column",
+            category=sqlalchemy.exc.SAWarning,
+        )
+
         url: URL = make_url(f"bigquery://{self.project_name!s}")
         if self.credentials_path:
             url = url.update_query_dict({"credentials_path": self.credentials_path})

airbyte/_processors/sql/duckdb.py

@@ -8,6 +8,7 @@
 from textwrap import dedent, indent
 from typing import TYPE_CHECKING, Union
 
+from duckdb_engine import DuckDBEngineWarning
 from overrides import overrides
 from pydantic import Field
 from typing_extensions import Literal
@@ -22,14 +23,6 @@
     from sqlalchemy.engine import Engine
 
 
-# Suppress warnings from DuckDB about reflection on indices.
-# https://github.com/Mause/duckdb_engine/issues/905
-warnings.filterwarnings(
-    "ignore",
-    message="duckdb-engine doesn't yet support reflection on indices",
-)
-
-
 # @dataclass
 class DuckDBConfig(SqlConfig):
     """Configuration for DuckDB."""
@@ -47,7 +40,13 @@ class DuckDBConfig(SqlConfig):
     @overrides
     def get_sql_alchemy_url(self) -> SecretString:
         """Return the SQLAlchemy URL to use."""
-        # return f"duckdb:///{self.db_path}?schema={self.schema_name}"
+        # Suppress warnings from DuckDB about reflection on indices.
+        # https://github.com/Mause/duckdb_engine/issues/905
+        warnings.filterwarnings(
+            "ignore",
+            message="duckdb-engine doesn't yet support reflection on indices",
+            category=DuckDBEngineWarning,
+        )
         return SecretString(f"duckdb:///{self.db_path!s}")
 
     @overrides

airbyte/_processors/sql/motherduck.py

@@ -6,6 +6,7 @@
 import warnings
 from typing import TYPE_CHECKING
 
+from duckdb_engine import DuckDBEngineWarning
 from overrides import overrides
 
 from airbyte._processors.file import JsonlWriter
@@ -21,6 +22,7 @@
 warnings.filterwarnings(
     "ignore",
     message="duckdb-engine doesn't yet support reflection on indices",
+    category=DuckDBEngineWarning,
 )
 
 

airbyte/_util/temp_files.py

@@ -5,6 +5,8 @@
 
 import json
 import tempfile
+import time
+import warnings
 from contextlib import contextmanager, suppress
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -20,18 +22,38 @@ def as_temp_files(files_contents: list[dict | str]) -> Generator[list[str], Any,
     temp_files: list[Any] = []
     try:
         for content in files_contents:
+            use_json = isinstance(content, dict)
             temp_file = tempfile.NamedTemporaryFile(
                 mode="w+t",
                 delete=False,
                 encoding="utf-8",
+                suffix=".json" if use_json else ".txt",
             )
             temp_file.write(
                 json.dumps(content) if isinstance(content, dict) else content,
             )
             temp_file.flush()
+            # Don't close the file yet (breaks Windows)
+            # temp_file.close()
             temp_files.append(temp_file)
         yield [file.name for file in temp_files]
     finally:
         for temp_file in temp_files:
-            with suppress(Exception):
-                Path(temp_file.name).unlink()
+            max_attempts = 5
+            for attempt in range(max_attempts):
+                try:
+                    with suppress(Exception):
+                        temp_file.close()
+
+                    Path(temp_file.name).unlink(missing_ok=True)
+
+                    break  # File was deleted successfully. Move on.
+                except Exception as ex:
+                    if attempt < max_attempts - 1:
+                        time.sleep(1)  # File might not be closed yet. Wait and try again.
+                    else:
+                        # Something went wrong and the file could not be deleted. Warn the user.
+                        warnings.warn(
+                            f"Failed to remove temporary file: '{temp_file.name}'. {ex}",
+                            stacklevel=2,
+                        )

airbyte/caches/duckdb.py

@@ -18,6 +18,7 @@
 
 import warnings
 
+from duckdb_engine import DuckDBEngineWarning
 from pydantic import PrivateAttr
 
 from airbyte._processors.sql.duckdb import DuckDBConfig, DuckDBSqlProcessor
@@ -29,6 +30,7 @@
 warnings.filterwarnings(
     "ignore",
     message="duckdb-engine doesn't yet support reflection on indices",
+    category=DuckDBEngineWarning,
 )
 
 

airbyte/caches/motherduck.py

@@ -16,6 +16,9 @@
 
 from __future__ import annotations
 
+import warnings
+
+from duckdb_engine import DuckDBEngineWarning
 from overrides import overrides
 from pydantic import Field, PrivateAttr
 
@@ -35,6 +38,14 @@ class MotherDuckConfig(DuckDBConfig):
     @overrides
     def get_sql_alchemy_url(self) -> SecretString:
         """Return the SQLAlchemy URL to use."""
+        # Suppress warnings from DuckDB about reflection on indices.
+        # https://github.com/Mause/duckdb_engine/issues/905
+        warnings.filterwarnings(
+            "ignore",
+            message="duckdb-engine doesn't yet support reflection on indices",
+            category=DuckDBEngineWarning,
+        )
+
         return SecretString(
             f"duckdb:///md:{self.database}?motherduck_token={self.api_key}"
             # f"&schema={self.schema_name}"  # TODO: Debug why this doesn't work

airbyte/cloud/experimental.py

@@ -27,6 +27,7 @@
 
 import warnings
 
+from airbyte import exceptions as exc
 from airbyte.cloud.connections import CloudConnection as Stable_CloudConnection
 from airbyte.cloud.workspaces import CloudWorkspace as Stable_CloudWorkspace
 
@@ -35,7 +36,7 @@
 # explicitly imports it.
 warnings.warn(
     message="The `airbyte.cloud.experimental` module is experimental and may change in the future.",
-    category=FutureWarning,
+    category=exc.AirbyteExperimentalFeatureWarning,
     stacklevel=2,
 )
 

airbyte/cloud/sync_results.py

@@ -294,8 +294,16 @@ def get_dataset(self, stream_name: str) -> CachedDataset:
         """Retrieve an `airbyte.datasets.CachedDataset` object for a given stream name.
 
         This can be used to read and analyze the data in a SQL-based destination.
+
+        TODO: In a future iteration, we can consider providing stream configuration information
+              (catalog information) to the `CachedDataset` object via the "Get stream properties"
+              API: https://reference.airbyte.com/reference/getstreamproperties
         """
-        return CachedDataset(self.get_sql_cache(), stream_name=stream_name)
+        return CachedDataset(
+            self.get_sql_cache(),
+            stream_name=stream_name,
+            stream_configuration=False,  # Don't look for stream configuration in cache.
+        )
 
     def get_sql_database_name(self) -> str:
         """Return the SQL database name."""

airbyte/datasets/_sql.py

@@ -139,10 +139,17 @@ def __init__(
         self,
         cache: CacheBase,
         stream_name: str,
+        stream_configuration: ConfiguredAirbyteStream | None | Literal[False] = None,
     ) -> None:
         """We construct the query statement by selecting all columns from the table.
 
         This prevents the need to scan the table schema to construct the query statement.
+
+        If stream_configuration is None, we attempt to retrieve the stream configuration from the
+        cache processor. This is useful when constructing a dataset from a CachedDataset object,
+        which already has the stream configuration.
+
+        If stream_configuration is set to False, we skip the stream configuration retrieval.
         """
         table_name = cache.processor.get_sql_table_name(stream_name)
         schema_name = cache.schema_name
@@ -151,6 +158,7 @@ def __init__(
             cache=cache,
             stream_name=stream_name,
             query_statement=query,
+            stream_configuration=stream_configuration,
         )
 
     @overrides

airbyte/exceptions.py

@@ -397,3 +397,10 @@ class AirbyteMultipleResourcesError(AirbyteError):
 
     resource_type: str | None = None
     resource_name_or_id: str | None = None
+
+
+# Custom Warnings
+
+
+class AirbyteExperimentalFeatureWarning(FutureWarning):
+    """Warning whenever using experimental features in PyAirbyte."""

airbyte/validate.py

@@ -63,7 +63,7 @@ def full_tests(connector_name: str, sample_config: str) -> None:
     source = ab.get_source(
         # TODO: FIXME: noqa: SIM115, PTH123
         connector_name,
-        config=json.load(open(sample_config, encoding="utf-8")),  # noqa: SIM115, PTH123,
+        config=json.loads(Path(sample_config).read_text(encoding="utf-8")),  # ,
         install_if_missing=False,
     )
 
@@ -118,8 +118,7 @@ def run() -> None:
 def validate(connector_dir: str, sample_config: str, *, validate_install_only: bool) -> None:
     # read metadata.yaml
     metadata_path = Path(connector_dir) / "metadata.yaml"
-    with Path(metadata_path).open(encoding="utf-8") as stream:
-        metadata = yaml.safe_load(stream)["data"]
+    metadata = yaml.safe_load(Path(metadata_path).read_text(encoding="utf-8"))["data"]
 
     # TODO: Use remoteRegistries.pypi.packageName once set for connectors
     connector_name = metadata["dockerRepository"].replace("airbyte/", "")
@@ -147,15 +146,24 @@ def validate(connector_dir: str, sample_config: str, *, validate_install_only: b
         ],
     }
 
-    with tempfile.NamedTemporaryFile(mode="w+t", delete=True, encoding="utf-8") as temp_file:
+    with tempfile.NamedTemporaryFile(
+        mode="w+t", delete=True, encoding="utf-8", suffix="-catalog.json"
+    ) as temp_file:
         temp_file.write(json.dumps(registry))
-        temp_file.seek(0)
+        temp_file.flush()
         os.environ["AIRBYTE_LOCAL_REGISTRY"] = str(temp_file.name)
-        if validate_install_only:
-            install_only_test(connector_name)
-        else:
-            if not sample_config:
-                raise exc.PyAirbyteInputError(
-                    input_value="--sample-config is required without --validate-install-only set"
-                )
-            full_tests(connector_name, sample_config)
+        try:
+            if validate_install_only:
+                install_only_test(connector_name)
+            else:
+                if not sample_config:
+                    raise exc.PyAirbyteInputError(
+                        input_value=(
+                            "`--sample-config` is required when `--validate-install-only`"
+                            "is not set."
+                        )
+                    )
+                full_tests(connector_name, sample_config)
+        finally:
+            del os.environ["AIRBYTE_LOCAL_REGISTRY"]
+            temp_file.close()