refactor: simplify exception class with kw_only=True and remove post_init

Co-Authored-By: Aaron <AJ> Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/cli/airbyte_cdk/_secrets.py

@@ -134,7 +134,7 @@ def fetch(
     secret_count = 0
     failed_secrets = []
     failed_secret_urls = []
-    
+
     for secret in secrets:
         secret_file_path = _get_secret_filepath(
             secrets_dir=secrets_dir,
@@ -145,7 +145,7 @@ def fetch(
             client=client,
             file_path=secret_file_path,
         )
-        
+
         if error:
             secret_name = secret.name.split("/secrets/")[-1]  # Removes project prefix
             failed_secrets.append(secret_name)
@@ -161,7 +161,7 @@ def fetch(
             f"No secrets found for connector: '{connector_name}'",
             err=True,
         )
-        
+
     if failed_secrets:
         error_message = f"Failed to retrieve {len(failed_secrets)} secret(s)"
         click.echo(
@@ -302,31 +302,31 @@ def _write_secret_file(
     file_path: Path,
 ) -> Optional[str]:
     """Write the most recent enabled version of a secret to a file.
-    
+
     Lists all enabled versions of the secret and selects the most recent one.
     Returns an error message if no enabled versions are found.
-    
+
     Args:
         secret: The secret to write to a file
         client: The Secret Manager client
         file_path: The path to write the secret to
-        
+
     Returns:
         Optional[str]: Error message if no enabled version is found, None otherwise
     """
     # List all enabled versions of the secret
     response = client.list_secret_versions(
         request={"parent": secret.name, "filter": "state:ENABLED"}
     )
-    
+
     versions = list(response)
-    
+
     if not versions:
         secret_name = secret.name.split("/secrets/")[-1]  # Removes project prefix
         return f"No enabled version found for secret: {secret_name}"
-    
+
     enabled_version = versions[0]
-    
+
     response = client.access_secret_version(name=enabled_version.name)
     file_path.write_text(response.payload.data.decode("UTF-8"))
     file_path.chmod(0o600)  # default to owner read/write only

airbyte_cdk/cli/airbyte_cdk/exceptions.py

@@ -7,21 +7,13 @@
 from airbyte_cdk.sql.exceptions import AirbyteConnectorError
 
 
-@dataclass
+@dataclass(kw_only=True)
 class ConnectorSecretWithNoValidVersionsError(AirbyteConnectorError):
     """Error when a connector secret has no valid versions."""
 
     connector_name: str
-    secret_names: List[str] = None  # type: ignore
-    connector_secret_urls: List[str] = None  # type: ignore
-    
-    def __post_init__(self):
-        """Initialize default values for lists."""
-        super().__post_init__()
-        if self.secret_names is None:
-            self.secret_names = []
-        if self.connector_secret_urls is None:
-            self.connector_secret_urls = []
+    secret_names: List[str]
+    connector_secret_urls: List[str]
 
     def __str__(self) -> str:
         """Return a string representation of the exception."""

unit_tests/cli/airbyte_cdk/test_secrets.py

@@ -56,7 +56,9 @@ def test_write_secret_file_with_enabled_version(self, mock_client, mock_secret,
         # Verify that no error was returned
         assert result is None
 
-    def test_write_secret_file_with_no_enabled_versions(self, mock_client, mock_secret, mock_file_path):
+    def test_write_secret_file_with_no_enabled_versions(
+        self, mock_client, mock_secret, mock_file_path
+    ):
         # Mock list_secret_versions to return an empty list (no enabled versions)
         mock_client.list_secret_versions.return_value = []
 
@@ -85,91 +87,104 @@ def test_write_secret_file_with_no_enabled_versions(self, mock_client, mock_secr
 @patch("airbyte_cdk.cli.airbyte_cdk._secrets._fetch_secret_handles")
 class TestFetch:
     def test_fetch_with_some_failed_secrets(
-        self, mock_fetch_secret_handles, mock_get_secrets_dir, mock_resolve, mock_get_client, tmp_path
+        self,
+        mock_fetch_secret_handles,
+        mock_get_secrets_dir,
+        mock_resolve,
+        mock_get_client,
+        tmp_path,
     ):
         # Setup mocks
         mock_client = MagicMock()
         mock_get_client.return_value = mock_client
-        
+
         mock_resolve.return_value = ("test-connector", tmp_path)
-        
+
         secrets_dir = tmp_path / "secrets"
         mock_get_secrets_dir.return_value = secrets_dir
-        
+
         # Create two secrets, one that will succeed and one that will fail
         secret1 = MagicMock()
         secret1.name = "projects/test-project/secrets/test-secret-1"
         secret1.labels = {}
-        
+
         secret2 = MagicMock()
         secret2.name = "projects/test-project/secrets/test-secret-2"
         secret2.labels = {}
-        
+
         mock_fetch_secret_handles.return_value = [secret1, secret2]
-        
+
         # Mock _write_secret_file to succeed for secret1 and fail for secret2
-        with patch("airbyte_cdk.cli.airbyte_cdk._secrets._write_secret_file") as mock_write_secret_file:
+        with patch(
+            "airbyte_cdk.cli.airbyte_cdk._secrets._write_secret_file"
+        ) as mock_write_secret_file:
             mock_write_secret_file.side_effect = [
                 None,  # Success for secret1
                 "No enabled version found for secret: test-secret-2",  # Failure for secret2
             ]
-            
+
             # Call the function
             runner = CliRunner()
             result = runner.invoke(fetch)
-            
+
             # Verify that _write_secret_file was called twice
             assert mock_write_secret_file.call_count == 2
-            
+
             # Verify that the error message was printed
             assert "Failed to retrieve secret 'test-secret-2'" in result.output
             assert "Failed to retrieve 1 secret(s)" in result.output
-            
+
             # Verify that the function did not raise an exception
             assert result.exit_code == 0
-    
+
     def test_fetch_with_all_failed_secrets(
-        self, mock_fetch_secret_handles, mock_get_secrets_dir, mock_resolve, mock_get_client, tmp_path
+        self,
+        mock_fetch_secret_handles,
+        mock_get_secrets_dir,
+        mock_resolve,
+        mock_get_client,
+        tmp_path,
     ):
         # Setup mocks
         mock_client = MagicMock()
         mock_get_client.return_value = mock_client
-        
+
         mock_resolve.return_value = ("test-connector", tmp_path)
-        
+
         secrets_dir = tmp_path / "secrets"
         mock_get_secrets_dir.return_value = secrets_dir
-        
+
         # Create two secrets that will both fail
         secret1 = MagicMock()
         secret1.name = "projects/test-project/secrets/test-secret-1"
         secret1.labels = {}
-        
+
         secret2 = MagicMock()
         secret2.name = "projects/test-project/secrets/test-secret-2"
         secret2.labels = {}
-        
+
         mock_fetch_secret_handles.return_value = [secret1, secret2]
-        
+
         # Mock _write_secret_file to fail for both secrets
-        with patch("airbyte_cdk.cli.airbyte_cdk._secrets._write_secret_file") as mock_write_secret_file:
+        with patch(
+            "airbyte_cdk.cli.airbyte_cdk._secrets._write_secret_file"
+        ) as mock_write_secret_file:
             mock_write_secret_file.side_effect = [
                 "No enabled version found for secret: test-secret-1",  # Failure for secret1
                 "No enabled version found for secret: test-secret-2",  # Failure for secret2
             ]
-            
+
             # Call the function
             runner = CliRunner()
             result = runner.invoke(fetch)
-            
+
             # Verify that _write_secret_file was called twice
             assert mock_write_secret_file.call_count == 2
-            
+
             # Verify that the error message was printed
             assert "Failed to retrieve secret 'test-secret-1'" in result.output
             assert "Failed to retrieve secret 'test-secret-2'" in result.output
             assert "Failed to retrieve 2 secret(s)" in result.output
-            
+
             # Verify that the function raised an exception
             assert result.exit_code != 0
-