Coverage for asyncer/_compat.py: 100%

1# AnyIO 4.1.0 renamed cancellable to abandon_on_cancel 

2import importlib 13 ctx1abcdefghijklm

3import importlib.metadata 13 ctx1abcdefghijklm

4from collections.abc import Callable 13 ctx1abcdefghijklm

5from typing import TypeVar 13 ctx1abcdefghijklm

6 

7import anyio 13 ctx1abcdefghijklm

8import anyio.to_thread 13 ctx1abcdefghijklm

9from anyio import CapacityLimiter 13 ctx1abcdefghijklm

10from typing_extensions import TypeVarTuple, Unpack 13 ctx1abcdefghijklm

11 

12ANYIO_VERSION = importlib.metadata.version("anyio") 13 ctx1abcdefghijklm

13 

14T_Retval = TypeVar("T_Retval") 13 ctx1abcdefghijklm

15PosArgsT = TypeVarTuple("PosArgsT") 13 ctx1abcdefghijklm

16 

17if ANYIO_VERSION >= "4.1.0": 13 ctx1abcdefghijklm

18 

19 async def run_sync( 11 ctx1abcdefghklm

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( 66 ctx1nopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!#$%'()*+,-./:;=?

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

27 ) 

28else: 

29 

30 async def run_sync( 2 ctx1ij

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 ctx2@ [ ] ^ _ ` { | } ~ abbb

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

38 )