feat: Add Python version compatibility checking during connector installation

airbyte/_executors/python.py

@@ -5,24 +5,68 @@
 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()
+
+    with suppress(Exception):
+        response = requests.get(
+            url=url,
+            headers={"User-Agent": f"PyAirbyte/{version}" if version else "PyAirbyte"},
+            timeout=10,
+        )
+
+        if not response.ok:
+            return None
+
+        data = response.json()
+        if not data:
+            return None
+        return data.get("info", {}).get("requires_python")
+
+    return None
+
+
 class VenvExecutor(Executor):
     def __init__(
         self,
@@ -106,6 +150,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 +245,40 @@ def get_installed_version(
 
             return None
 
+    def _check_python_version_compatibility(
+        self,
+        package_name: str,
+        requires_python: str | None,
+    ) -> bool | None:
+        """Check if current Python version is compatible with package requirements.
+
+        Returns True if confirmed, False if incompatible, or None if no determination can be made.
+
+        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 None
+
+        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,
+            )
+            return False
+        return True
+
     def ensure_installation(
         self,
         *,

pyproject.toml

@@ -29,6 +29,7 @@ google-cloud-secret-manager = "^2.17.0"
 jsonschema = ">=3.2.0,<5.0"
 orjson = "^3.10"
 overrides = "^7.4.0"
+packaging = "^23.0"
 pandas = { version = ">=1.5.3,<3.0" }
 psycopg = {extras = ["binary", "pool"], version = "^3.1.19"}
 psycopg2-binary =  "^2.9.9"

tests/unit_tests/test_python_executor.py

@@ -0,0 +1,186 @@
+from unittest.mock import Mock, patch
+import requests
+
+from airbyte._executors.python import VenvExecutor, _get_pypi_python_requirements_cached
+
+
+class TestGetPypiPythonRequirementsCached:
+    """Test the _get_pypi_python_requirements_cached function."""
+
+    def test_offline_mode_returns_none(self):
+        """Test that offline mode returns None without making network calls."""
+        with patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", True):
+            _get_pypi_python_requirements_cached.cache_clear()
+            result = _get_pypi_python_requirements_cached("airbyte-source-hubspot")
+            assert result is None
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_successful_api_response(self, mock_get):
+        """Test successful PyPI API response returns requires_python."""
+        mock_response = Mock()
+        mock_response.ok = True
+        mock_response.json.return_value = {"info": {"requires_python": "<3.12,>=3.10"}}
+        mock_get.return_value = mock_response
+
+        _get_pypi_python_requirements_cached.cache_clear()
+        result = _get_pypi_python_requirements_cached("airbyte-source-hubspot")
+
+        assert result == "<3.12,>=3.10"
+        mock_get.assert_called_once()
+        call_args = mock_get.call_args
+        assert (
+            call_args[1]["url"] == "https://pypi.org/pypi/airbyte-source-hubspot/json"
+        )
+        assert call_args[1]["timeout"] == 10
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_package_not_found_returns_none(self, mock_get):
+        """Test that 404 response returns None."""
+        mock_response = Mock()
+        mock_response.ok = False
+        mock_get.return_value = mock_response
+
+        _get_pypi_python_requirements_cached.cache_clear()
+        result = _get_pypi_python_requirements_cached("nonexistent-package")
+
+        assert result is None
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_json_parsing_error_returns_none(self, mock_get):
+        """Test that JSON parsing errors return None."""
+        mock_response = Mock()
+        mock_response.ok = True
+        mock_response.json.side_effect = ValueError("Invalid JSON")
+        mock_get.return_value = mock_response
+
+        _get_pypi_python_requirements_cached.cache_clear()
+        result = _get_pypi_python_requirements_cached("test-package")
+
+        assert result is None
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_missing_requires_python_field_returns_none(self, mock_get):
+        """Test that missing requires_python field returns None."""
+        mock_response = Mock()
+        mock_response.ok = True
+        mock_response.json.return_value = {"info": {"name": "test-package"}}
+        mock_get.return_value = mock_response
+
+        _get_pypi_python_requirements_cached.cache_clear()
+        result = _get_pypi_python_requirements_cached("test-package")
+
+        assert result is None
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_network_timeout_returns_none(self, mock_get):
+        """Test that network timeouts return None."""
+        mock_get.side_effect = requests.exceptions.Timeout()
+
+        _get_pypi_python_requirements_cached.cache_clear()
+        result = _get_pypi_python_requirements_cached("test-package")
+
+        assert result is None
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_caching_behavior(self, mock_get):
+        """Test that lru_cache works correctly."""
+        mock_response = Mock()
+        mock_response.ok = True
+        mock_response.json.return_value = {"info": {"requires_python": ">=3.8"}}
+        mock_get.return_value = mock_response
+
+        _get_pypi_python_requirements_cached.cache_clear()
+
+        result1 = _get_pypi_python_requirements_cached("test-package")
+        result2 = _get_pypi_python_requirements_cached("test-package")
+
+        assert result1 == ">=3.8"
+        assert result2 == ">=3.8"
+        mock_get.assert_called_once()
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_user_agent_header(self, mock_get):
+        """Test that proper User-Agent header is sent."""
+        mock_response = Mock()
+        mock_response.ok = True
+        mock_response.json.return_value = {"info": {"requires_python": ">=3.8"}}
+        mock_get.return_value = mock_response
+
+        with patch("airbyte._executors.python.get_version", return_value="1.0.0"):
+            _get_pypi_python_requirements_cached.cache_clear()
+            _get_pypi_python_requirements_cached("test-package")
+
+        call_args = mock_get.call_args
+        assert "PyAirbyte/1.0.0" in call_args[1]["headers"]["User-Agent"]
+
+    @patch("airbyte._executors.python.AIRBYTE_OFFLINE_MODE", False)
+    @patch("requests.get")
+    def test_user_agent_header_no_version(self, mock_get):
+        """Test User-Agent header when version is None."""
+        mock_response = Mock()
+        mock_response.ok = True
+        mock_response.json.return_value = {"info": {"requires_python": ">=3.8"}}
+        mock_get.return_value = mock_response
+
+        with patch("airbyte._executors.python.get_version", return_value=None):
+            _get_pypi_python_requirements_cached.cache_clear()
+            _get_pypi_python_requirements_cached("test-package")
+
+        call_args = mock_get.call_args
+        assert call_args[1]["headers"]["User-Agent"] == "PyAirbyte"
+
+
+class TestVenvExecutorVersionCompatibility:
+    """Test version compatibility checking in VenvExecutor."""
+
+    def test_check_python_version_compatibility_no_requirements(self):
+        """Test that None requirements return None."""
+        executor = VenvExecutor(name="test-source")
+        result = executor._check_python_version_compatibility("test-package", None)
+        assert result is None
+
+    @patch("airbyte._executors.python.warn_once")
+    def test_check_python_version_compatibility_incompatible(self, mock_warn):
+        """Test warning for incompatible Python version."""
+        executor = VenvExecutor(name="test-source")
+
+        mock_version_info = Mock()
+        mock_version_info.major = 3
+        mock_version_info.minor = 13
+        mock_version_info.micro = 0
+
+        with patch("sys.version_info", mock_version_info):
+            result = executor._check_python_version_compatibility(
+                "test-package", "<3.12,>=3.10"
+            )
+
+        assert result is False
+        mock_warn.assert_called_once()
+        warning_message = mock_warn.call_args[0][0]
+        assert "Python version compatibility warning" in warning_message
+        assert "test-package" in warning_message
+        assert "3.13.0" in warning_message
+        assert "<3.12,>=3.10" in warning_message
+
+    def test_check_python_version_compatibility_compatible(self):
+        """Test compatible Python version returns True."""
+        executor = VenvExecutor(name="test-source")
+
+        mock_version_info = Mock()
+        mock_version_info.major = 3
+        mock_version_info.minor = 11
+        mock_version_info.micro = 5
+
+        with patch("sys.version_info", mock_version_info):
+            result = executor._check_python_version_compatibility(
+                "test-package", "<3.12,>=3.10"
+            )
+
+        assert result is True