feat: validate_parameters for GSheets (#15578)

* feat: validate_parameters for GSheets

* Move import inside fixture

* Update deps

* Rename parameter

superset/db_engine_specs/gsheets.py

@@ -17,14 +17,17 @@
 import json
 import re
 from contextlib import closing
-from typing import Any, Dict, Optional, Pattern, Tuple, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, Pattern, Tuple, TYPE_CHECKING
 
+from flask import g
 from flask_babel import gettext as __
+from sqlalchemy.engine import create_engine
 from sqlalchemy.engine.url import URL
+from typing_extensions import TypedDict
 
 from superset import security_manager
 from superset.db_engine_specs.sqlite import SqliteEngineSpec
-from superset.errors import SupersetErrorType
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 
 if TYPE_CHECKING:
     from superset.models.core import Database
@@ -33,6 +36,12 @@ if TYPE_CHECKING:
 SYNTAX_ERROR_REGEX = re.compile('SQLError: near "(?P<server_error>.*?)": syntax error')
 
 
+class GSheetsParametersType(TypedDict):
+    credentials_info: Dict[str, Any]
+    query: Dict[str, Any]
+    table_catalog: Dict[str, str]
+
+
 class GSheetsEngineSpec(SqliteEngineSpec):
     """Engine for Google spreadsheets"""
 
@@ -45,7 +54,7 @@ class GSheetsEngineSpec(SqliteEngineSpec):
         SYNTAX_ERROR_REGEX: (
             __(
                 'Please check your query for syntax errors near "%(server_error)s". '
-                "Then, try running your query again."
+                "Then, try running your query again.",
             ),
             SupersetErrorType.SYNTAX_ERROR,
             {},
@@ -54,7 +63,7 @@ class GSheetsEngineSpec(SqliteEngineSpec):
 
     @classmethod
     def modify_url_for_impersonation(
-        cls, url: URL, impersonate_user: bool, username: Optional[str]
+        cls, url: URL, impersonate_user: bool, username: Optional[str],
     ) -> None:
         if impersonate_user and username is not None:
             user = security_manager.find_user(username=username)
@@ -77,3 +86,41 @@ class GSheetsEngineSpec(SqliteEngineSpec):
             metadata = {}
 
         return {"metadata": metadata["extra"]}
+
+    @classmethod
+    def validate_parameters(
+        cls, parameters: GSheetsParametersType,
+    ) -> List[SupersetError]:
+        errors: List[SupersetError] = []
+
+        credentials_info = parameters.get("credentials_info")
+        table_catalog = parameters.get("table_catalog", {})
+
+        if not table_catalog:
+            return errors
+
+        # We need a subject in case domain wide delegation is set, otherwise the
+        # check will fail. This means that the admin will be able to add sheets
+        # that only they have access, even if later users are not able to access
+        # them.
+        subject = g.user.email if g.user else None
+
+        engine = create_engine(
+            "gsheets://", service_account_info=credentials_info, subject=subject,
+        )
+        conn = engine.connect()
+        for name, url in table_catalog.items():
+            try:
+                results = conn.execute(f'SELECT * FROM "{url}" LIMIT 1')
+                results.fetchall()
+            except Exception:  # pylint: disable=broad-except
+                errors.append(
+                    SupersetError(
+                        message=f"Unable to connect to spreadsheet {name} at {url}",
+                        error_type=SupersetErrorType.TABLE_DOES_NOT_EXIST_ERROR,
+                        level=ErrorLevel.WARNING,
+                        extra={"name": name, "url": url},
+                    ),
+                )
+
+        return errors