[MAINTENANCE] Add check to CloudDataContext to ensure using latest PyPI version

great_expectations/data_context/_version_checker.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import json
+import logging
+
+import requests
+from packaging import version
+from typing_extensions import TypedDict
+
+logger = logging.getLogger()
+
+
+class _PyPIPackageInfo(TypedDict):
+    version: str
+
+
+class _PyPIPackageData(TypedDict):
+    info: _PyPIPackageInfo
+
+
+class _VersionChecker:
+
+    _BASE_PYPI_URL = "https://pypi.org/pypi"
+    _PYPI_GX_ENDPOINT = f"{_BASE_PYPI_URL}/great_expectations/json"
+
+    def __init__(self, user_version: str) -> None:
+        self._user_version = version.Version(user_version)
+
+    def check_if_using_latest_gx(self) -> None:
+        pypi_version = self._get_latest_version_from_pypi()
+        if not pypi_version:
+            logger.debug("Could not compare with latest PyPI version; skipping check.")
+            return
+
+        if self._is_using_outdated_release(pypi_version):
+            self._warn_user(pypi_version)
+
+    def _get_latest_version_from_pypi(self) -> version.Version | None:
+        response_json: _PyPIPackageData | None = None
+        try:
+            response = requests.get(self._PYPI_GX_ENDPOINT)
+            response.raise_for_status()
+            response_json = response.json()
+        except json.JSONDecodeError as jsonError:
+            logger.debug(f"Failed to parse PyPI API response into JSON: {jsonError}")
+        except requests.HTTPError as http_err:
+            logger.debug(
+                f"An HTTP error occurred when trying to hit PyPI API: {http_err}"
+            )
+        except requests.Timeout as timeout_exc:
+            logger.debug(
+                f"Failed to hit the PyPI API due a timeout error: {timeout_exc}"
+            )
+
+        if not response_json:
+            return None
+
+        info = response_json["info"]
+        pkg_version = info["version"]
+
+        return version.Version(pkg_version)
+
+    def _is_using_outdated_release(self, pypi_version: version.Version) -> bool:
+        return pypi_version > self._user_version
+
+    def _warn_user(self, pypi_version: version.Version) -> None:
+        logger.warning(
+            f"You are using great_expectations version {self._user_version}; "
+            f"however, version {pypi_version} is available.\nYou should consider "
+            "upgrading via `pip install great_expectations --upgrade`\n."
+        )

great_expectations/data_context/data_context/cloud_data_context.py

@@ -26,6 +26,7 @@
     _ConfigurationProvider,
 )
 from great_expectations.core.serializer import JsonConfigSerializer
+from great_expectations.data_context._version_checker import _VersionChecker
 from great_expectations.data_context.cloud_constants import (
     CLOUD_DEFAULT_BASE_URL,
     GXCloudEnvironmentVariable,
@@ -103,6 +104,7 @@ def __init__(
             ge_cloud_organization_id=ge_cloud_organization_id,
         )
 
+        self._check_if_latest_version()
         self._cloud_config = self.get_cloud_config(
             cloud_base_url=cloud_base_url,
             cloud_access_token=cloud_access_token,
@@ -118,6 +120,10 @@ def __init__(
             runtime_environment=runtime_environment,
         )
 
+    def _check_if_latest_version(self) -> None:
+        checker = _VersionChecker(__version__)
+        checker.check_if_using_latest_gx()
+
     def _init_project_config(
         self, project_config: Optional[Union[DataContextConfig, Mapping]]
     ) -> DataContextConfig: