Coverage for asyncer / _compat.py: 100%

1# AnyIO 4.1.0 renamed cancellable to abandon_on_cancel 

2import importlib 7 ctx1abcdefg

3import importlib.metadata 7 ctx1abcdefg

4from collections.abc import Callable 7 ctx1abcdefg

5from typing import TypeVar 7 ctx1abcdefg

6 

7import anyio 7 ctx1abcdefg

8import anyio.to_thread 7 ctx1abcdefg

9from anyio import CapacityLimiter 7 ctx1abcdefg

10from typing_extensions import TypeVarTuple, Unpack 7 ctx1abcdefg

11 

12ANYIO_VERSION = importlib.metadata.version("anyio") 7 ctx1abcdefg

13 

14T_Retval = TypeVar("T_Retval") 7 ctx1abcdefg

15PosArgsT = TypeVarTuple("PosArgsT") 7 ctx1abcdefg

16 

17if ANYIO_VERSION >= "4.1.0": 7 ctx1abcdefg

18 

19 async def run_sync( 5 ctx1abcdg

20 func: Callable[[Unpack[PosArgsT]], T_Retval], 

21 *args: Unpack[PosArgsT], 

22 abandon_on_cancel: bool = False, 

23 limiter: CapacityLimiter | None = None, 

24 ) -> T_Retval: 

25 return await anyio.to_thread.run_sync( 30 ctx1hijklmnopqrstuvwxyzABCDEFGHIJK

26 func, *args, abandon_on_cancel=abandon_on_cancel, limiter=limiter 

27 ) 

28else: 

29 

30 async def run_sync( 2 ctx1ef

31 func: Callable[[Unpack[PosArgsT]], T_Retval], 

32 *args: Unpack[PosArgsT], 

33 abandon_on_cancel: bool = False, 

34 limiter: CapacityLimiter | None = None, 

35 ) -> T_Retval: 

36 return await anyio.to_thread.run_sync( 12 ctx1LMNOPQRSTUVW

37 func, *args, cancellable=abandon_on_cancel, limiter=limiter 

38 )