cover anyio’s NoEventLoopError without altering existing behavior

[gotsysdba]

Fixes #144

Note once the minimal version of anyio ≥ 4.11, the backward compatibility code can be removed:

---

# anyio.NoEventLoopError was added in anyio 4.11. anyio 4.10 is still supported
# (via the langgraph/evaluation extras), where the attribute does not exist.
# Once the minimum anyio version is ≥ 4.11, replace this with a plain
# tuple and catch anyio.NoEventLoopError directly (commented out below)
_NO_LOOP_ERRORS: tuple[type[BaseException], ...] = (AsyncLibraryNotFoundError,)
if hasattr(anyio, "NoEventLoopError"):
    _NO_LOOP_ERRORS = (*_NO_LOOP_ERRORS, anyio.NoEventLoopError)


def get_execution_context() -> AsyncContext:
    """
    Return one of:
    - 'sync'         → plain synchronous context (no loop, no worker thread)
    - 'sync_worker'  → synchronous worker thread (spawned by to_thread.run_sync)
    - 'async'        → running inside the event loop
    """
    try:
        anyio.get_current_task()
        return AsyncContext.ASYNC
    # except (AsyncLibraryNotFoundError, anyio.NoEventLoopError): # anyio>4.10
    except _NO_LOOP_ERRORS:

Replace with:

def get_execution_context() -> AsyncContext:
    """
    Return one of:
    - 'sync'         → plain synchronous context (no loop, no worker thread)
    - 'sync_worker'  → synchronous worker thread (spawned by to_thread.run_sync)
    - 'async'        → running inside the event loop
    """
    try:
        anyio.get_current_task()
        return AsyncContext.ASYNC
    except (AsyncLibraryNotFoundError, anyio.NoEventLoopError):

[paul-cayet]

@gotsysdba , I tried your solution using

• anyio: 4.12.1
• sniffio: 1.3.1

But I was still getting an error

File ".../anyio/_core/_eventloop.py", line 187, in get_async_backend
    raise NoCurrentAsyncBackend
anyio._core._eventloop.NoCurrentAsyncBackend: Not currently running on any asynchronous event loopAvailable async backends: asyncio, trio

Using the following works for me however:

def _is_anyio_worker_thread() -> bool:
    try:
        # AnyIO "claims" worker threads spawned by `to_thread.run_sync()` by storing
        # a current_token on this thread-local. Plain synchronous threads do not have
        # it, and AnyIO's own `from_thread.run*()` helpers use the same signal, so
        # this is more robust than checking thread names or version-specific
        # "no event loop" exceptions.
        from_thread.threadlocals.current_token  # type: ignore[attr-defined]
    except AttributeError:
        return False
    else:
        return True


def get_execution_context() -> AsyncContext:
    """
    Return one of:
    - 'sync'         → plain synchronous context (no loop, no worker thread)
    - 'sync_worker'  → synchronous worker thread (spawned by to_thread.run_sync)
    - 'async'        → running inside the event loop
    """
    try:
        current_async_library()
        return AsyncContext.ASYNC
    except AsyncLibraryNotFoundError:
        if _is_anyio_worker_thread():
            # for anyio workers, we can use specific methods to
            # handle back asynchronous code to the main loop
            return AsyncContext.SYNC_WORKER

        # otherwise, consider it as a synchronous thread
        return AsyncContext.SYNC

[gotsysdba]

Thanks @paul-cayet, yours is a much cleaner, more reliable detection though I couldn't find any docs on from_thread.threadlocals which leads me to believe it is a private API?... I did find docs on check_cancelled (https://anyio.readthedocs.io/en/stable/threads.html) which will only succeed if inside a worker thread. So suggesting a minor API change (pushed).

I must have tested mine back on MacOS (where it wasn't a problem) and/or with my monkeypatch in place.. too many terminals :(). I've tested this suggestion on both MacOS and OL8.

TYVM for the review.

[dhilloulinoracle]

Internal regression succeeded 🍏: Build ID #394