chore(compat): add CURRENT_PY_VERSION and bump NEXT_PY_VERSION to 3.16

Centralize the 3.15 wiring bound and the 3.16 fail-closed bound so wrap()
can keep raising on 3.15 while NEXT_PY means first unsupported.

Author: vlad-scherbich

ddtrace/internal/compat.py

@@ -11,6 +11,8 @@
 
 __all__ = [
     "maybe_stringify",
+    "CURRENT_PY_VERSION",
+    "CURRENT_PY_VERSION_INFO",
     "NEXT_PY_UNSUPPORTED_MSG",
     "NEXT_PY_VERSION",
     "NEXT_PY_VERSION_INFO",
@@ -19,10 +21,19 @@
 
 PYTHON_VERSION_INFO = sys.version_info
 
-# First CPython version that wrapping / bytecode injection do not support yet.
-NEXT_PY_VERSION: str = "3.15"
-_next_py_parts = NEXT_PY_VERSION.split(".")[:2]
-NEXT_PY_VERSION_INFO: tuple[int, int] = (int(_next_py_parts[0]), int(_next_py_parts[1]))
+
+def _py_version_info(version: str) -> tuple[int, int]:
+    major, minor = version.split(".")[:2]
+    return (int(major), int(minor))
+
+
+# CPython version currently being wired.
+CURRENT_PY_VERSION: str = "3.15"
+CURRENT_PY_VERSION_INFO: tuple[int, int] = _py_version_info(CURRENT_PY_VERSION)
+
+# First unsupported CPython (exclusive upper bound / fail-closed default).
+NEXT_PY_VERSION: str = "3.16"
+NEXT_PY_VERSION_INFO: tuple[int, int] = _py_version_info(NEXT_PY_VERSION)
 NEXT_PY_UNSUPPORTED_MSG: str = "This version of CPython is not supported yet: {}.{}".format(*sys.version_info[:2])
 
 

ddtrace/internal/coverage/instrumentation_py3_12.py

@@ -16,6 +16,7 @@
 from bytecode import Bytecode
 
 from ddtrace.internal.bytecode_injection import HookType
+from ddtrace.internal.compat import CURRENT_PY_VERSION_INFO
 from ddtrace.internal.coverage.import_instrumentation_py3_12 import ImportName
 from ddtrace.internal.coverage.import_instrumentation_py3_12 import ImportNamesByLine
 from ddtrace.internal.coverage.import_instrumentation_py3_12 import import_names_by_line
@@ -44,7 +45,7 @@
 # In Python 3.15 (PEP 810 lazy imports), IMPORT_NAME's arg is bit-packed:
 # bits 2+ = name index into co_names, bits 0-1 = lazy/eager flags.
 # So the index is arg >> 2. On 3.12-3.14, arg is a plain index (shift by 0).
-_IMPORT_NAME_ARG_SHIFT = 2 if sys.version_info >= (3, 15) else 0
+_IMPORT_NAME_ARG_SHIFT = 2 if sys.version_info >= CURRENT_PY_VERSION_INFO else 0
 
 # Detect empty modules: the bytecode pattern varies across Python versions.
 # Python 3.12-3.13: RESUME + RETURN_CONST
@@ -56,7 +57,7 @@
 # Check if file-level coverage is requested
 _USE_FILE_LEVEL_COVERAGE = asbool(env.get("_DD_COVERAGE_FILE_LEVEL", "true"))
 _ACCURATE_IMPORTS_REQUESTED = asbool(env.get("_DD_COVERAGE_ACCURATE_IMPORTS", "false"))
-_USE_ACCURATE_IMPORTS = sys.version_info < (3, 15) and _ACCURATE_IMPORTS_REQUESTED
+_USE_ACCURATE_IMPORTS = sys.version_info < CURRENT_PY_VERSION_INFO and _ACCURATE_IMPORTS_REQUESTED
 if _ACCURATE_IMPORTS_REQUESTED and not _USE_ACCURATE_IMPORTS:
     log.info(
         "_DD_COVERAGE_ACCURATE_IMPORTS is enabled, but accurate import tracking is not supported on Python %s; "

ddtrace/internal/module.py

@@ -829,18 +829,18 @@ def _trace(frame, event, arg):
 
 
 def lazy(f: t.Callable[[], None]) -> None:
-    from ddtrace.internal.compat import NEXT_PY_VERSION_INFO
+    from ddtrace.internal.compat import CURRENT_PY_VERSION_INFO
     from ddtrace.internal.compat import PYTHON_VERSION_INFO
 
     _globals = sys._getframe(1).f_globals
     _initialized = False
 
-    if PYTHON_VERSION_INFO < NEXT_PY_VERSION_INFO:
+    if PYTHON_VERSION_INFO < CURRENT_PY_VERSION_INFO:
         LazyWrappingContext(t.cast(FunctionType, f)).wrap()
 
     def __getattr__(name: str) -> t.Any:
         nonlocal _initialized
-        if PYTHON_VERSION_INFO >= NEXT_PY_VERSION_INFO:
+        if PYTHON_VERSION_INFO >= CURRENT_PY_VERSION_INFO:
             if not _initialized:
                 _exec_lazy_init(t.cast(FunctionType, f), _globals)
                 _initialized = True

ddtrace/internal/monitoring.py

@@ -21,12 +21,14 @@
 from typing import Optional
 import weakref
 
+from ddtrace.internal.compat import CURRENT_PY_VERSION
+from ddtrace.internal.compat import CURRENT_PY_VERSION_INFO
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.threads import Lock
 
 
-if sys.version_info < (3, 15):
-    raise ImportError("ddtrace.internal.monitoring requires Python 3.15+")
+if sys.version_info < CURRENT_PY_VERSION_INFO:
+    raise ImportError("ddtrace.internal.monitoring requires Python %s+" % CURRENT_PY_VERSION)
 
 log = get_logger(__name__)
 

ddtrace/internal/wrapping/__init__.py

@@ -13,8 +13,8 @@
 from bytecode import Instr
 
 from ddtrace.internal.assembly import Assembly
+from ddtrace.internal.compat import CURRENT_PY_VERSION_INFO
 from ddtrace.internal.compat import NEXT_PY_UNSUPPORTED_MSG
-from ddtrace.internal.compat import NEXT_PY_VERSION_INFO
 from ddtrace.internal.threads import Lock
 from ddtrace.internal.wrapping.asyncs import wrap_async
 from ddtrace.internal.wrapping.generators import wrap_generator
@@ -300,7 +300,8 @@ def wrap_bytecode(wrapper: Wrapper, wrapped: FunctionType) -> bc.Bytecode:
     return a coroutine function, and so on. The signature is also preserved to
     avoid breaking, e.g., usages of the ``inspect`` module.
     """
-    if PY >= NEXT_PY_VERSION_INFO:
+    # Still fail-closed on CURRENT_PY_VERSION; NEXT_PY_VERSION is the exclusive upper bound.
+    if PY >= CURRENT_PY_VERSION_INFO:
         raise NotImplementedError(NEXT_PY_UNSUPPORTED_MSG)
 
     code = wrapped.__code__
@@ -349,7 +350,7 @@ def wrap(f: FunctionType, wrapper: Wrapper) -> WrappedFunction:
     Note that this changes the behavior of the original function with the
     wrapper function, instead of creating a new function object.
     """
-    if PY >= NEXT_PY_VERSION_INFO:
+    if PY >= CURRENT_PY_VERSION_INFO:
         raise NotImplementedError(NEXT_PY_UNSUPPORTED_MSG)
     wrapped = FunctionType(
         code := f.__code__,

lib-injection/sources/sitecustomize.py

@@ -43,6 +43,7 @@ def parse_version(version):
 RUNTIMES_ALLOW_LIST = {
     "cpython": {
         "min": Version(version=(3, 9), constraint=""),
+        # Exclusive upper bound; keep in sync with NEXT_PY_VERSION in ddtrace.internal.compat.
         "max": Version(version=(3, 16), constraint=""),
     }
 }

tests/internal/test_py315_import_degrade.py

@@ -9,6 +9,8 @@
 
 import pytest
 
+from ddtrace.internal.compat import CURRENT_PY_VERSION
+from ddtrace.internal.compat import CURRENT_PY_VERSION_INFO
 from ddtrace.internal.compat import NEXT_PY_UNSUPPORTED_MSG
 from ddtrace.internal.compat import NEXT_PY_VERSION
 from ddtrace.internal.compat import NEXT_PY_VERSION_INFO
@@ -23,6 +25,13 @@ def test_unsupported_msg_includes_running_version():
     assert NEXT_PY_UNSUPPORTED_MSG == _UNSUPPORTED_MSG
 
 
+def test_current_and_next_py_version_constants():
+    assert CURRENT_PY_VERSION == "3.15"
+    assert CURRENT_PY_VERSION_INFO == (3, 15)
+    assert NEXT_PY_VERSION == "3.16"
+    assert NEXT_PY_VERSION_INFO == (3, 16)
+
+
 def test_wrapping_modules_import():
     import ddtrace.internal.bytecode_injection  # noqa: F401
     import ddtrace.internal.module  # noqa: F401
@@ -31,7 +40,7 @@ def test_wrapping_modules_import():
     import ddtrace.internal.wrapping.generators  # noqa: F401
 
 
-@pytest.mark.skipif(PYTHON_VERSION_INFO < NEXT_PY_VERSION_INFO, reason=f"{NEXT_PY_VERSION} wrap() degrade")
+@pytest.mark.skipif(PYTHON_VERSION_INFO < CURRENT_PY_VERSION_INFO, reason=f"{CURRENT_PY_VERSION} wrap() degrade")
 def test_wrap_raises_not_implemented_on_315():
     from ddtrace.internal.wrapping import wrap
 
@@ -45,7 +54,7 @@ def wrapper(wrapped, args, kwargs):  # noqa: ANN001, ANN202
         wrap(f, wrapper)
 
 
-@pytest.mark.skipif(PYTHON_VERSION_INFO < NEXT_PY_VERSION_INFO, reason=f"{NEXT_PY_VERSION} lazy module degrade")
+@pytest.mark.skipif(PYTHON_VERSION_INFO < CURRENT_PY_VERSION_INFO, reason=f"{CURRENT_PY_VERSION} lazy module degrade")
 def test_lazy_module_decorator_without_bytecode_wrap():
     import tests.internal.lazy as lazy_module
 
@@ -62,7 +71,9 @@ def test_exec_lazy_init_without_source():
     assert module_globals["exported"] == 123
 
 
-@pytest.mark.skipif(PYTHON_VERSION_INFO < NEXT_PY_VERSION_INFO, reason=f"{NEXT_PY_VERSION} debugging products degrade")
+@pytest.mark.skipif(
+    PYTHON_VERSION_INFO < CURRENT_PY_VERSION_INFO, reason=f"{CURRENT_PY_VERSION} debugging products degrade"
+)
 def test_debugging_products_load_without_failure():
     from ddtrace.internal.products import ProductManager
 
@@ -77,7 +88,7 @@ def test_debugging_products_load_without_failure():
         assert product_name not in product_manager._failed
 
 
-@pytest.mark.skipif(PYTHON_VERSION_INFO < NEXT_PY_VERSION_INFO, reason=f"{NEXT_PY_VERSION} inject_hook degrade")
+@pytest.mark.skipif(PYTHON_VERSION_INFO < CURRENT_PY_VERSION_INFO, reason=f"{CURRENT_PY_VERSION} inject_hook degrade")
 def test_inject_hook_raises_not_implemented_on_315():
     from ddtrace.internal.bytecode_injection import inject_hook