fix: keep litellm importable on Python 3.10 and guard 3.11-only typing imports in CI (#39448)

* ci: guard against Python 3.10-incompatible typing imports

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* ci: address Python 3.10 typing guard review

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(ci): honor version-guard direction and scan litellm-proxy-extras in py310 typing check

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yassin <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

.github/workflows/test-code-quality.yml

@@ -128,6 +128,9 @@ jobs:
       - name: check_fastuuid_usage
         run: uv run --no-sync python ./tests/code_coverage_tests/check_fastuuid_usage.py
 
+      - name: check_py310_typing_imports
+        run: uv run --no-sync python ./tests/code_coverage_tests/check_py310_typing_imports.py
+
       - name: check_e2e_no_raw_requests
         run: uv run --no-sync python ./tests/code_coverage_tests/check_e2e_no_raw_requests.py
 
@@ -145,3 +148,33 @@ jobs:
 
       - name: documentation_test_api_docs
         run: uv run --no-sync python ./tests/documentation_tests/test_api_docs.py
+
+  python-310-import-smoke:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.10"
+
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Install dependencies
+        run: uv sync --frozen --extra proxy --python 3.10
+
+      - run: uv run --no-sync python --version
+
+      - name: Import litellm
+        run: uv run --no-sync python -c "import litellm"
+
+      - name: Check litellm CLI
+        run: uv run --no-sync litellm --version

litellm/integrations/websearch_interception/handler.py

@@ -10,9 +10,9 @@ import asyncio
 import math
 import uuid
 from collections.abc import AsyncIterator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Final, Literal, Never, TypedDict, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict, TypeVar, cast
 
-from typing_extensions import ReadOnly
+from typing_extensions import Never, ReadOnly
 
 import litellm
 from litellm._logging import verbose_logger

litellm/proxy/_experimental/mcp_server/tool_search.py

@@ -5,10 +5,10 @@ from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Final, TypedDict, assert_never
+from typing import TYPE_CHECKING, Any, Final, TypedDict
 
 from pydantic import ValidationError
-from typing_extensions import ReadOnly, Required
+from typing_extensions import ReadOnly, Required, assert_never
 
 import litellm
 from litellm.proxy.agent_endpoints.agent_search import DEFAULT_AGENT_SEARCH_TOP_K

litellm/proxy/agent_endpoints/endpoints.py

@@ -13,10 +13,10 @@ import os
 import uuid
 from collections.abc import Mapping, Sequence
 from types import MappingProxyType
-from typing import Annotated, Final, TypedDict, assert_never
+from typing import Annotated, Final, TypedDict
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
-from typing_extensions import ReadOnly, Required
+from typing_extensions import ReadOnly, Required, assert_never
 
 import litellm
 from litellm._logging import verbose_proxy_logger

litellm/proxy/common_utils/reset_budget_job.py

@@ -7,7 +7,9 @@ from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 from types import MappingProxyType
-from typing import Final, Literal, Protocol, TypeVar, assert_never
+from typing import Final, Literal, Protocol, TypeVar
+
+from typing_extensions import assert_never
 
 import litellm
 from litellm._logging import verbose_proxy_logger

litellm/proxy/openai_files_endpoints/batch_file_validation.py

@@ -2,7 +2,9 @@ import json
 from collections.abc import Iterator
 from dataclasses import dataclass
 from itertools import chain
-from typing import BinaryIO, Final, NoReturn, assert_never
+from typing import BinaryIO, Final, NoReturn
+
+from typing_extensions import assert_never
 
 from litellm.proxy._types import ProxyException
 

litellm/proxy/openai_files_endpoints/general_upload_validation.py

@@ -8,7 +8,9 @@ extensions, path-traversal filenames) regardless of purpose.
 
 from dataclasses import dataclass
 from pathlib import Path
-from typing import BinaryIO, Final, NoReturn, assert_never
+from typing import BinaryIO, Final, NoReturn
+
+from typing_extensions import assert_never
 
 from litellm.proxy._types import ProxyException
 from litellm.proxy.common_utils.path_utils import safe_filename

litellm/types/llms/gemini_audio_transcription.py

@@ -1,7 +1,7 @@
-from typing import Literal, Required
+from typing import Literal
 
 from pydantic import BaseModel, ConfigDict
-from typing_extensions import ReadOnly, TypedDict
+from typing_extensions import ReadOnly, Required, TypedDict
 
 
 class GeminiTranscriptionAudioInput(TypedDict):

tests/code_coverage_tests/check_py310_typing_imports.py

@@ -0,0 +1,150 @@
+import ast
+import os
+import sys
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+PY311_PLUS_TYPING_NAMES: Final[frozenset[str]] = frozenset(
+    {
+        "NotRequired",
+        "Required",
+        "Self",
+        "LiteralString",
+        "Never",
+        "assert_never",
+        "assert_type",
+        "reveal_type",
+        "TypeVarTuple",
+        "Unpack",
+        "dataclass_transform",
+        "override",
+        "TypeAliasType",
+        "get_original_bases",
+        "ReadOnly",
+        "TypeIs",
+        "NoDefault",
+        "get_protocol_members",
+        "is_protocol",
+        "evaluate_forward_ref",
+        "TypeForm",
+    }
+)
+
+
+@dataclass(frozen=True, slots=True)
+class TypingImportViolation:
+    file: str
+    line: int
+    name: str
+
+
+def _walk_with_ancestors(
+    node: ast.AST, ancestors: tuple[tuple[ast.AST, str], ...] = ()
+) -> Iterator[tuple[ast.AST, tuple[tuple[ast.AST, str], ...]]]:
+    yield node, ancestors
+    for field_name, field_value in ast.iter_fields(node):
+        if isinstance(field_value, ast.AST):
+            yield from _walk_with_ancestors(field_value, (*ancestors, (node, field_name)))
+        elif isinstance(field_value, list):
+            for child in field_value:
+                if isinstance(child, ast.AST):
+                    yield from _walk_with_ancestors(child, (*ancestors, (node, field_name)))
+
+
+def _is_sys_version_info(node: ast.AST) -> bool:
+    return (
+        isinstance(node, ast.Attribute)
+        and isinstance(node.value, ast.Name)
+        and node.value.id == "sys"
+        and node.attr == "version_info"
+    )
+
+
+def _is_version_guarded(ancestors: tuple[tuple[ast.AST, str], ...]) -> bool:
+    nearest_if: Final[tuple[ast.If, str] | None] = next(
+        (
+            (ancestor, field_name)
+            for ancestor, field_name in reversed(ancestors)
+            if isinstance(ancestor, ast.If)
+        ),
+        None,
+    )
+    if nearest_if is None:
+        return False
+    enclosing_if, branch = nearest_if
+    test: Final[ast.expr] = enclosing_if.test
+    if not isinstance(test, ast.Compare) or len(test.ops) != 1 or not _is_sys_version_info(test.left):
+        return False
+    operator: Final[ast.cmpop] = test.ops[0]
+    return (isinstance(operator, (ast.Gt, ast.GtE)) and branch == "body") or (
+        isinstance(operator, (ast.Lt, ast.LtE)) and branch == "orelse"
+    )
+
+
+def scan_file(file_path: str | os.PathLike[str]) -> tuple[TypingImportViolation, ...]:
+    path: Final[Path] = Path(file_path)
+    tree: Final[ast.Module] = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    return tuple(
+        violation
+        for node, ancestors in _walk_with_ancestors(tree)
+        if not _is_version_guarded(ancestors)
+        for violation in _violations_for_node(node, path)
+    )
+
+
+def _violations_for_node(
+    node: ast.AST, path: Path
+) -> tuple[TypingImportViolation, ...]:
+    if isinstance(node, ast.ImportFrom) and node.module == "typing":
+        return tuple(
+            TypingImportViolation(file=str(path), line=node.lineno, name=alias.name)
+            for alias in node.names
+            if alias.name in PY311_PLUS_TYPING_NAMES
+        )
+    if (
+        isinstance(node, ast.Attribute)
+        and isinstance(node.value, ast.Name)
+        and node.value.id == "typing"
+        and node.attr in PY311_PLUS_TYPING_NAMES
+    ):
+        return (TypingImportViolation(file=str(path), line=node.lineno, name=node.attr),)
+    return ()
+
+
+def scan_directory(base_dir: str | os.PathLike[str] = ".") -> tuple[TypingImportViolation, ...]:
+    base_path: Final[Path] = Path(base_dir)
+    return tuple(
+        violation
+        for directory in (
+            base_path / "litellm",
+            base_path / "enterprise",
+            base_path / "litellm-proxy-extras" / "litellm_proxy_extras",
+        )
+        if directory.exists()
+        for path in directory.rglob("*.py")
+        for violation in scan_file(path)
+    )
+
+
+def main() -> None:
+    violations: Final[tuple[TypingImportViolation, ...]] = scan_directory()
+    if violations:
+        message: Final[str] = "\n".join(
+            (
+                "Python 3.10-incompatible typing imports found:",
+                *(
+                    f"{violation.file}:{violation.line}: {violation.name} is unavailable in Python 3.10; "
+                    "import it from typing_extensions instead because litellm supports Python 3.10"
+                    for violation in violations
+                ),
+            )
+        )
+        sys.stdout.write(f"{message}\n")
+        raise RuntimeError("Import Python 3.10-incompatible typing names from typing_extensions instead")
+    sys.stdout.write("No Python 3.10-incompatible typing imports found.\n")
+
+
+if __name__ == "__main__":
+    main()

tests/test_litellm/test_check_py310_typing_imports.py

@@ -0,0 +1,86 @@
+import sys
+from pathlib import Path
+from typing import Final
+
+_CODE_COVERAGE_DIR: Final[Path] = Path(__file__).resolve().parents[1] / "code_coverage_tests"
+sys.path.insert(0, str(_CODE_COVERAGE_DIR))  # test-quality-ok: required to import checker from its source directory
+import check_py310_typing_imports as checker  # noqa: E402  # load checker from its source directory
+
+
+def _scan(tmp_path: Path, source: str) -> tuple[object, ...]:
+    file_path = tmp_path / "fixture.py"
+    file_path.write_text(source, encoding="utf-8")
+    return checker.scan_file(file_path)
+
+
+def test_typing_import_flags_python_311_name(tmp_path: Path) -> None:
+    violations = _scan(tmp_path, "from typing import NotRequired, TypedDict\n")
+    assert tuple(violation.name for violation in violations) == ("NotRequired",)
+
+
+def test_typing_extensions_import_passes(tmp_path: Path) -> None:
+    assert _scan(tmp_path, "from typing_extensions import NotRequired\n") == ()
+
+
+def test_typing_attribute_flags_python_311_name(tmp_path: Path) -> None:
+    violations = _scan(tmp_path, "import typing\nx: typing.Self\n")
+    assert tuple(violation.name for violation in violations) == ("Self",)
+
+
+def test_version_guarded_typing_import_passes(tmp_path: Path) -> None:
+    source = (
+        "import sys\n"
+        "if sys.version_info >= (3, 11):\n"
+        "    from typing import NotRequired\n"
+        "else:\n"
+        "    from typing_extensions import NotRequired\n"
+    )
+    assert _scan(tmp_path, source) == ()
+
+
+def test_python_310_branch_flags_typing_import(tmp_path: Path) -> None:
+    source = (
+        "import sys\n"
+        "if sys.version_info >= (3, 11):\n"
+        "    from typing_extensions import NotRequired\n"
+        "else:\n"
+        "    from typing import NotRequired\n"
+    )
+    violations = _scan(tmp_path, source)
+    assert tuple(violation.name for violation in violations) == ("NotRequired",)
+
+
+def test_python_310_branch_is_exempt_for_less_than_guard(tmp_path: Path) -> None:
+    source = (
+        "import sys\n"
+        "if sys.version_info < (3, 11):\n"
+        "    from typing_extensions import NotRequired\n"
+        "else:\n"
+        "    from typing import NotRequired\n"
+    )
+    assert _scan(tmp_path, source) == ()
+
+
+def test_nearest_if_controls_version_guard(tmp_path: Path) -> None:
+    source = (
+        "if sys.version_info >= (3, 11):\n"
+        "    from typing import Self\n"
+        "    x = 1\n"
+        "if True:\n"
+        "    from typing import Self\n"
+    )
+    violations = _scan(tmp_path, source)
+    assert tuple((violation.name, violation.line) for violation in violations) == (("Self", 5),)
+
+
+def test_scan_directory_includes_proxy_extras(tmp_path: Path) -> None:
+    file_path = tmp_path / "litellm-proxy-extras" / "litellm_proxy_extras" / "m.py"
+    file_path.parent.mkdir(parents=True)
+    file_path.write_text("from typing import NotRequired\n", encoding="utf-8")
+
+    violations = checker.scan_directory(tmp_path)
+    assert tuple((violation.name, violation.file) for violation in violations) == (("NotRequired", str(file_path)),)
+
+
+def test_python_310_typing_name_passes(tmp_path: Path) -> None:
+    assert _scan(tmp_path, "from typing import Optional\n") == ()