feat: Add Python version compatibility checking during connector installation

airbyte/_executors/python.py

@@ -5,24 +5,69 @@
 import subprocess
 import sys
 from contextlib import suppress
+from functools import lru_cache
 from pathlib import Path
 from shutil import rmtree
 from typing import TYPE_CHECKING, Literal
 
+import requests
 from overrides import overrides
+from packaging.specifiers import SpecifierSet
+from packaging.version import Version
 from rich import print  # noqa: A004  # Allow shadowing the built-in
 
 from airbyte import exceptions as exc
 from airbyte._executors.base import Executor
 from airbyte._util.meta import is_windows
 from airbyte._util.telemetry import EventState, log_install_state
 from airbyte._util.venv_util import get_bin_dir
+from airbyte.constants import AIRBYTE_OFFLINE_MODE
+from airbyte.logs import warn_once
+from airbyte.version import get_version
 
 
 if TYPE_CHECKING:
     from airbyte.sources.registry import ConnectorMetadata
 
 
+@lru_cache(maxsize=128)
+def _get_pypi_python_requirements_cached(package_name: str) -> str | None:
+    """Get the requires_python field from PyPI for a package.
+
+    Args:
+        package_name: The PyPI package name to check
+
+    Returns:
+        The requires_python string from PyPI, or None if unavailable
+
+    Example:
+        For airbyte-source-hubspot, returns "<3.12,>=3.10"
+    """
+    if AIRBYTE_OFFLINE_MODE:
+        return None
+
+    url = f"https://pypi.org/pypi/{package_name}/json"
+    version = get_version()
+    response = requests.get(
+        url=url,
+        headers={"User-Agent": f"PyAirbyte/{version}" if version else "PyAirbyte"},
+        timeout=10,
+    )
+
+    if not response.ok:
+        return None
+
+    try:
+        data = response.json()
+        return data.get("info", {}).get("requires_python")
+    except Exception:
+        # Intentionally broad exception handling to ensure silent failure in all scenarios:
+        # - JSON parsing errors (json.JSONDecodeError)
+        # - Any other unexpected errors
+        # This ensures connector installation never fails due to version checking issues.
+        return None
+
+
 class VenvExecutor(Executor):
     def __init__(
         self,
@@ -106,6 +151,14 @@ def install(self) -> None:
 
         After installation, the installed version will be stored in self.reported_version.
         """
+        package_name = (
+            self.metadata.pypi_package_name
+            if self.metadata and self.metadata.pypi_package_name
+            else f"airbyte-{self.name}"
+        )
+        requires_python = _get_pypi_python_requirements_cached(package_name)
+        self._check_python_version_compatibility(package_name, requires_python)
+
         self._run_subprocess_and_raise_on_failure(
             [sys.executable, "-m", "venv", str(self._get_venv_path())]
         )
@@ -193,6 +246,39 @@ def get_installed_version(
 
             return None
 
+    def _check_python_version_compatibility(
+        self,
+        package_name: str,
+        requires_python: str | None,
+    ) -> None:
+        """Check if current Python version is compatible with package requirements.
+
+        Args:
+            package_name: Name of the package being checked
+            requires_python: The requires_python constraint from PyPI (e.g., "<3.12,>=3.10")
+        """
+        if not requires_python:
+            return
+
+        try:
+            current_version = (
+                f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
+            )
+
+            spec_set = SpecifierSet(requires_python)
+            current_ver = Version(current_version)
+
+            if current_ver not in spec_set:
+                warn_once(
+                    f"Python version compatibility warning for '{package_name}': "
+                    f"Current Python {current_version} may not be compatible with "
+                    f"package requirement '{requires_python}'. "
+                    f"Installation will proceed but may fail.",
+                    with_stack=False,
+                )
+        except Exception:
+            pass
+
     def ensure_installation(
         self,
         *,