Replace bytecode inspection with co_names and move sniffio to runtime deps

Author: Nikhil172913832

pyproject.toml

@@ -43,6 +43,7 @@ dependencies = [
   # Pact dependencies
   "pact-python-ffi~=0.4.0",
   # External dependencies
+  "sniffio~=1.0",
   "yarl~=1.0",
   "typing-extensions~=4.0 ; python_version < '3.13'",
 ]
@@ -120,7 +121,6 @@ test = [
   "pytest-rerunfailures~=16.0",
   "pytest~=9.0",
   "requests~=2.0",
-  "sniffio>=1.3",
   "testcontainers~=4.0",
 ]
 types = [

src/pact/_util.py

@@ -11,7 +11,6 @@
 
 import asyncio
 import contextvars
-import dis
 import inspect
 import logging
 import socket
@@ -26,15 +25,13 @@
 
 try:
     import sniffio  # type: ignore[import-not-found]
-except ImportError:
+except ImportError:  # pragma: no cover
     sniffio = None  # type: ignore[assignment]
 
 try:
     import trio  # type: ignore[import-not-found]
-    from trio.lowlevel import current_trio_token  # type: ignore[import-not-found]
-except ImportError:
+except ImportError:  # pragma: no cover
     trio = None  # type: ignore[assignment]
-    current_trio_token = None  # type: ignore[assignment]
 
 try:
     import curio  # type: ignore[import-not-found,import-untyped]
@@ -223,7 +220,6 @@ def apply_args(f: Callable[..., _T], args: Mapping[str, object]) -> _T:  # noqa:
     Returns:
         The result of the function.
     """
-    # Check if f is a partial wrapping an async function
     func_to_check = f.func if isinstance(f, partial) else f
     is_async = inspect.iscoroutinefunction(func_to_check)
     signature = inspect.signature(f)
@@ -252,8 +248,6 @@ def apply_args(f: Callable[..., _T], args: Mapping[str, object]) -> _T:  # noqa:
     # First, we inspect the keyword arguments and try and pass in some arguments
     # by currying them in.
     for param in signature.parameters.values():
-        # Try matching the parameter name, or if it starts with underscore,
-        # also try matching without the leading underscore.
         arg_key = None
         if param.name in args:
             arg_key = param.name
@@ -320,25 +314,20 @@ def apply_args(f: Callable[..., _T], args: Mapping[str, object]) -> _T:  # noqa:
                 },
             )
 
-    try:
-        if is_async:
-            result = f()
-            if inspect.iscoroutine(result):
-                return _run_async_coroutine(result)
-            return result
-        return f()
-    except Exception:
-        logger.exception("Error occurred while calling function %s", f_name)
-        raise
-
-
-def _run_async_coroutine(coro: Coroutine[Any, Any, _T]) -> _T:  # noqa: C901
+    if is_async:
+        result = f()
+        if inspect.iscoroutine(result):
+            return _run_async_coroutine(result)
+        return result  # pragma: no cover
+    return f()
+
+
+def _run_async_coroutine(coro: Coroutine[Any, Any, _T]) -> _T:
     """
     Run a coroutine in an event loop.
 
-    Detects the async runtime (asyncio, trio, or curio) and executes the
-    coroutine appropriately. Preserves ContextVars when creating a new event
-    loop, which is important when handlers are called from threads.
+    Detects the current async runtime and runs the coroutine in it,
+    preserving ContextVars across the dispatch.
 
     Args:
         coro:
@@ -351,24 +340,13 @@ def _run_async_coroutine(coro: Coroutine[Any, Any, _T]) -> _T:  # noqa: C901
         RuntimeError:
             If the detected runtime (trio or curio) is not installed.
     """
-    runtime = _detect_async_runtime_from_coroutine(coro)
+    runtime = _detect_async_runtime(coro)
 
     if runtime == "trio":
         if trio is None:
             msg = "trio is not installed"
             raise RuntimeError(msg)
 
-        if current_trio_token is not None:
-            try:
-                token = current_trio_token()
-
-                async def _run_with_token() -> _T:
-                    return await coro
-
-                return trio.from_thread.run_sync(_run_with_token, trio_token=token)  # type: ignore[return-value]
-            except RuntimeError:
-                pass
-
         ctx = contextvars.copy_context()
 
         async def _run_trio() -> _T:
@@ -404,13 +382,17 @@ async def _run_curio() -> _T:
     return ctx.run(asyncio.run, coro)  # type: ignore[arg-type,return-value]
 
 
-def _detect_async_runtime_from_coroutine(coro: Coroutine[Any, Any, _T]) -> str:  # noqa: C901
+def _detect_async_runtime(coro: Coroutine[Any, Any, _T]) -> str:
     """
-    Detect async runtime by inspecting the coroutine object.
+    Detect the async runtime to use for a given coroutine.
+
+    When called from within a running async context, `sniffio` is used to
+    identify the library. Otherwise the coroutine's `co_names` is inspected
+    for `trio` or `curio` references.
 
     Args:
         coro:
-            The coroutine object to inspect.
+            The coroutine to inspect.
 
     Returns:
         The detected runtime: "asyncio", "trio", or "curio".
@@ -421,47 +403,10 @@ def _detect_async_runtime_from_coroutine(coro: Coroutine[Any, Any, _T]) -> str:
         except sniffio.AsyncLibraryNotFoundError:
             pass
 
-    # Inspect bytecode to check for qualified attribute access (e.g., trio.sleep)
-    # This is more robust than just checking co_names for module and method separately
-    func_code = coro.cr_code  # type: ignore[attr-defined]
-
-    # Parse bytecode to find LOAD_GLOBAL/LOAD_NAME followed by LOAD_ATTR patterns
-    # This detects qualified accesses like `trio.sleep()` or `curio.spawn()`
-    bytecode = list(dis.get_instructions(func_code))
-
-    trio_detected = False
-    curio_detected = False
-
-    for i, instr in enumerate(bytecode):
-        # Check for module.attribute pattern (LOAD_GLOBAL/LOAD_NAME + LOAD_ATTR)
-        if instr.opname in ("LOAD_GLOBAL", "LOAD_NAME") and i + 1 < len(bytecode):
-            next_instr = bytecode[i + 1]
-            if next_instr.opname == "LOAD_ATTR":
-                module_name = instr.argval
-                attr_name = next_instr.argval
-
-                # Check for trio-specific qualified access
-                if module_name == "trio":
-                    trio_indicators = {
-                        "sleep",
-                        "open_nursery",
-                        "CancelScope",
-                        "current_trio_token",
-                    }
-                    if attr_name in trio_indicators:
-                        trio_detected = True
-
-                # Check for curio-specific qualified access
-                elif module_name == "curio":
-                    curio_indicators = {"sleep", "spawn", "TaskGroup", "AWAIT"}
-                    if attr_name in curio_indicators:
-                        curio_detected = True
-
-    # Trio takes precedence if both are detected
-    if trio_detected:
+    names = set(coro.cr_code.co_names)  # type: ignore[attr-defined]
+    if trio is not None and "trio" in names:
         return "trio"
-    if curio_detected:
+    if curio is not None and "curio" in names:
         return "curio"
 
-    # Default to asyncio as it's the most common
     return "asyncio"