Coverage for src/chat_limiter/utils.py: 85%

1""" 

2Utilities for handling asyncio execution contexts. 

3 

4Fail-fast, minimal helpers to bridge async code into sync callers. 

5""" 

6 

7import asyncio 

8import concurrent.futures 

9from typing import Any, TypeVar 

10 

11T = TypeVar("T") 

12 

13 

14def run_coro_blocking(coro: Any) -> T: 

15 """Run the given coroutine to completion, regardless of event loop state. 

16 

17 - If no loop is running, call asyncio.run(coro). 

18 - If a loop is running (e.g., Jupyter), execute asyncio.run(coro) in a new thread. 

19 """ 

20 try: 

21 asyncio.get_running_loop() 

22 in_async_context = True 

23 except RuntimeError: 

24 in_async_context = False 

25 

26 if not in_async_context: 

27 try: 

28 return asyncio.run(coro) # type: ignore[no-any-return] 

29 finally: 

30 # Ensure the coroutine is closed even if asyncio.run was mocked 

31 # and did not consume it, to avoid "coroutine was never awaited" warnings. 

32 try: 

33 coro.close() # type: ignore[attr-defined] 

34 except Exception: 

35 pass 

36 

37 def _runner() -> T: 

38 return asyncio.run(coro) # type: ignore[no-any-return] 

39 

40 with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: 

41 try: 

42 future: concurrent.futures.Future[T] = executor.submit(_runner) 

43 return future.result() 

44 finally: 

45 # Close the coroutine in case the background runner did not 

46 # actually execute it (e.g., asyncio.run mocked in tests). 

47 try: 

48 coro.close() # type: ignore[attr-defined] 

49 except Exception: 

50 pass 

51 

52