fix(session): include root cause for invalid project, region, or session template

google/cloud/managed_spark_connect/session.py

@@ -52,6 +52,7 @@
     Aborted,
     FailedPrecondition,
     InvalidArgument,
+    MethodNotImplemented,
     NotFound,
     PermissionDenied,
 )
@@ -427,6 +428,118 @@ def _is_valid_session_id(session_id: str) -> bool:
     return bool(re.match(pattern, session_id))
 
 
+_SESSION_TEMPLATE_RESOURCE_PATTERN = re.compile(
+    r"^projects/[^/]+/(?:locations|regions)/[^/]+/sessionTemplates/[a-zA-Z0-9_-]+$"
+)
+_SESSION_TEMPLATE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
+
+
+def _is_valid_session_template_ref(template: str) -> bool:
+    if "/" in template:
+        return bool(_SESSION_TEMPLATE_RESOURCE_PATTERN.match(template))
+    return bool(_SESSION_TEMPLATE_ID_PATTERN.match(template))
+
+
+def _extract_api_error_details(
+    exc: BaseException,
+) -> Tuple[List[str], List[str]]:
+    """Return (reasons, debug_details) from a google.api_core GoogleAPICallError."""
+    reasons: List[str] = []
+    debug_details: List[str] = []
+    for item in getattr(exc, "details", None) or ():
+        reason = getattr(item, "reason", None)
+        if isinstance(reason, str) and reason and reason not in reasons:
+            reasons.append(reason)
+        detail = getattr(item, "detail", None)
+        if (
+            isinstance(detail, str)
+            and detail
+            and not detail.startswith("[ORIGINAL ERROR]")
+            and detail not in debug_details
+        ):
+            debug_details.append(detail)
+        for violation in getattr(item, "field_violations", None) or ():
+            field = getattr(violation, "field", "")
+            desc = getattr(violation, "description", "")
+            if desc:
+                entry = f"{field}: {desc}" if field else desc
+                if entry not in debug_details:
+                    debug_details.append(entry)
+    return reasons, debug_details
+
+
+def _format_create_session_error(
+    exc: BaseException,
+    project_id: Optional[str],
+    region: Optional[str],
+    api_endpoint: Optional[str],
+    session_template: Optional[str] = None,
+) -> str:
+    prefix = "Error while creating Managed Spark Session"
+    raw_msg = (getattr(exc, "message", None) or str(exc)).strip()
+    if isinstance(exc, MethodNotImplemented) and "status: 404" in raw_msg:
+        default_endpoint = (
+            f"{region}-dataproc.googleapis.com" if region else None
+        )
+        if region and (
+            api_endpoint is None or api_endpoint == default_endpoint
+        ):
+            return (
+                f"{prefix}: Regional endpoint '{default_endpoint}' "
+                f"(derived from location '{region}') returned HTTP 404; "
+                f"check that '{region}' is a valid Managed Spark region."
+            )
+        endpoint_desc = f" '{api_endpoint}'" if api_endpoint else ""
+        region_desc = f" (location '{region}')" if region else ""
+        return (
+            f"{prefix}: Endpoint{endpoint_desc}{region_desc} "
+            f"returned HTTP 404: {raw_msg}"
+        )
+
+    base_msg = raw_msg
+    reasons, debug_details = _extract_api_error_details(exc)
+    extra_parts: List[str] = []
+
+    for detail in debug_details:
+        if detail not in base_msg:
+            extra_parts.append(detail)
+
+    if not debug_details:
+        if "CONSUMER_INVALID" in reasons and project_id:
+            note = f"Project '{project_id}' was not found or is invalid."
+            if note not in base_msg:
+                extra_parts.append(note)
+        elif "RESOURCE_PROJECT_INVALID" in reasons and project_id:
+            note = f"Invalid project ID '{project_id}'."
+            if note not in base_msg:
+                extra_parts.append(note)
+        elif "LOCATION_POLICY_VIOLATED" in reasons and region:
+            note = f"Invalid location '{region}'."
+            if note not in base_msg:
+                extra_parts.append(note)
+
+    if (
+        isinstance(exc, InvalidArgument)
+        and session_template
+        and not _is_valid_session_template_ref(session_template)
+        and "session_template" not in base_msg
+        and "Session Template" not in base_msg
+        and "session template" not in base_msg
+    ):
+        extra_parts.append(
+            f"Invalid session_template '{session_template}'; expected a "
+            "template ID or 'projects/{project}/locations/{location}/"
+            "sessionTemplates/{template}'."
+        )
+
+    if base_msg and extra_parts and base_msg[-1] not in ".!?:;":
+        base_msg = f"{base_msg}."
+    parts = [p for p in (base_msg, *extra_parts) if p]
+    if not parts:
+        return prefix
+    return f"{prefix}: {' '.join(parts)}"
+
+
 def _timestamp_str(value: Any) -> Optional[str]:
     if value is None:
         return None
@@ -941,12 +1054,24 @@ def create_session_pbar():
                             logger.error(
                                 f"Exception while writing active session to file {file_path}, {e}"
                             )
-                except (InvalidArgument, PermissionDenied) as e:
+                except (
+                    InvalidArgument,
+                    PermissionDenied,
+                    MethodNotImplemented,
+                ) as e:
                     ManagedSparkSession._active_s8s_session_id = None
                     ManagedSparkSession._active_session_uses_custom_id = False
                     raise ManagedSparkConnectException(
-                        f"Error while creating Managed Spark Session: {e.message}"
-                    )
+                        _format_create_session_error(
+                            e,
+                            self._project_id,
+                            self._region,
+                            _get_client_option(
+                                self._client_options, "api_endpoint"
+                            ),
+                            session_config.session_template,
+                        )
+                    ) from e
                 except DefaultCredentialsError as e:
                     ManagedSparkSession._active_s8s_session_id = None
                     ManagedSparkSession._active_session_uses_custom_id = False

tests/integration/test_session.py

@@ -14,6 +14,11 @@
 import datetime
 import uuid
 
+from google.api_core.exceptions import (
+    InvalidArgument,
+    MethodNotImplemented,
+    PermissionDenied,
+)
 from google.cloud.dataproc_v1 import (
     CreateSessionTemplateRequest,
     DeleteSessionRequest,
@@ -25,6 +30,9 @@
     TerminateSessionRequest,
 )
 from google.cloud.managed_spark_connect import ManagedSparkSession
+from google.cloud.managed_spark_connect.exceptions import (
+    ManagedSparkConnectException,
+)
 from pyspark.errors.exceptions import connect as connect_exceptions
 from pyspark.sql.types import StringType
 import pytest
@@ -516,3 +524,114 @@ def test_serverless_session_termination_and_deletion_lifecycle(
                         )
                 except Exception:
                     pass
+
+
+def test_create_spark_session_with_invalid_project_id(
+    test_project, test_region, os_environment, isolated_session_state
+):
+    """Test that a nonexistent project ID raises a clear ManagedSparkConnectException."""
+    invalid_project = f"{test_project}-nonexistent-{uuid.uuid4().hex[:8]}"
+
+    with pytest.raises(ManagedSparkConnectException) as exc_info:
+        (
+            ManagedSparkSession.builder.projectId(invalid_project)
+            .location(test_region)
+            .getOrCreate()
+        )
+
+    assert isinstance(exc_info.value.__cause__, PermissionDenied)
+    msg = str(exc_info.value)
+    assert msg.startswith("Error while creating Managed Spark Session:")
+    assert f"Permission denied on resource project {invalid_project}." in msg
+    assert "not found" in msg
+    assert ManagedSparkSession._active_s8s_session_id is None
+
+
+def test_create_spark_session_with_invalid_region(
+    test_project, os_environment, isolated_session_state
+):
+    """Test that an invalid region raises a clear ManagedSparkConnectException explaining the 404."""
+    invalid_region = "us-east-4"
+
+    with pytest.raises(ManagedSparkConnectException) as exc_info:
+        (
+            ManagedSparkSession.builder.projectId(test_project)
+            .location(invalid_region)
+            .getOrCreate()
+        )
+
+    assert isinstance(exc_info.value.__cause__, MethodNotImplemented)
+    msg = str(exc_info.value)
+    assert msg == (
+        "Error while creating Managed Spark Session: Regional endpoint"
+        " 'us-east-4-dataproc.googleapis.com' (derived from location"
+        " 'us-east-4') returned HTTP 404; check that 'us-east-4' is a valid"
+        " Managed Spark region."
+    )
+    assert ManagedSparkSession._active_s8s_session_id is None
+
+
+def test_create_spark_session_with_invalid_session_template(
+    test_project, test_region, os_environment, isolated_session_state
+):
+    """Test that a nonexistent session template raises a clear ManagedSparkConnectException."""
+    template_id = f"nonexistent-template-{uuid.uuid4().hex[:8]}"
+    nonexistent_template = (
+        f"projects/{test_project}/locations/{test_region}/sessionTemplates/"
+        f"{template_id}"
+    )
+    session_config = Session()
+    session_config.session_template = nonexistent_template
+
+    with pytest.raises(ManagedSparkConnectException) as exc_info:
+        (
+            ManagedSparkSession.builder.projectId(test_project)
+            .location(test_region)
+            .sessionConfig(session_config)
+            .getOrCreate()
+        )
+
+    assert isinstance(exc_info.value.__cause__, InvalidArgument)
+    msg = str(exc_info.value)
+    assert msg.startswith("Error while creating Managed Spark Session:")
+    assert f"'{template_id}' does not exist" in msg
+    assert ManagedSparkSession._active_s8s_session_id is None
+
+    bare_template_config = Session()
+    bare_template_config.session_template = template_id
+
+    with pytest.raises(ManagedSparkConnectException) as bare_exc_info:
+        (
+            ManagedSparkSession.builder.projectId(test_project)
+            .location(test_region)
+            .sessionConfig(bare_template_config)
+            .getOrCreate()
+        )
+
+    assert isinstance(bare_exc_info.value.__cause__, InvalidArgument)
+    bare_msg = str(bare_exc_info.value)
+    assert bare_msg.startswith("Error while creating Managed Spark Session:")
+    assert f"'{template_id}' does not exist" in bare_msg
+    assert ManagedSparkSession._active_s8s_session_id is None
+
+    mismatched_region_config = Session()
+    mismatched_region_config.session_template = (
+        f"projects/{test_project}/locations/us-east-4/sessionTemplates/"
+        f"{template_id}"
+    )
+
+    with pytest.raises(ManagedSparkConnectException) as mismatch_exc_info:
+        (
+            ManagedSparkSession.builder.projectId(test_project)
+            .location(test_region)
+            .sessionConfig(mismatched_region_config)
+            .getOrCreate()
+        )
+
+    assert isinstance(mismatch_exc_info.value.__cause__, InvalidArgument)
+    mismatch_msg = str(mismatch_exc_info.value)
+    assert mismatch_msg.startswith(
+        "Error while creating Managed Spark Session:"
+    )
+    assert "us-east-4" in mismatch_msg
+    assert ManagedSparkSession._active_s8s_session_id is None

tests/unit/test_init.py

@@ -1251,7 +1251,9 @@ def _boom():
     print(json.dumps({
         "type": f"{type(e).__module__}.{type(e).__qualname__}",
         "message": str(e),
+        "msg": getattr(e, "msg", None),
         "is_import_error": isinstance(e, ImportError),
+        "is_runtime_error": isinstance(e, RuntimeError),
         "name": getattr(e, "name", None),
     }))
 """
@@ -1329,7 +1331,9 @@ def test_missing_pyspark_with_missing_or_broken_ipython_recommends_client(
     def test_missing_pyspark_is_an_import_error_named_pyspark(self):
         error = self._run("plain", "absent")
         self.assertTrue(error["is_import_error"])
+        self.assertFalse(error["is_runtime_error"])
         self.assertEqual(error["name"], "pyspark")
+        self.assertEqual(error["msg"], error["message"])
 
     def test_leftover_pyspark_namespace_dir_recommends_client(self):
         error = self._run("plain", "namespace")
@@ -1355,7 +1359,9 @@ def test_old_pyspark_recommends_upgrading_via_spark_extra(self):
                 # Caught by the same `except ImportError:` guards as a
                 # missing pyspark.
                 self.assertTrue(error["is_import_error"])
+                self.assertFalse(error["is_runtime_error"])
                 self.assertEqual(error["name"], "pyspark")
+                self.assertEqual(error["msg"], error["message"])
                 self.assertEqual(
                     error["message"],
                     "PySpark 3.5.1 is installed, but google-cloud-spark-connect "