Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiosqlite/context.py: 54%

1# Copyright Amethyst Reese 

2# Licensed under the MIT license 

3 

4 

5from collections.abc import Coroutine, Generator 

6from contextlib import AbstractAsyncContextManager 

7from functools import wraps 

8from typing import Any, Callable, TypeVar 

9 

10from .cursor import Cursor 

11 

12_T = TypeVar("_T") 

13 

14 

15class Result(AbstractAsyncContextManager[_T], Coroutine[Any, Any, _T]): 

16 __slots__ = ("_coro", "_obj") 

17 

18 def __init__(self, coro: Coroutine[Any, Any, _T]): 

19 self._coro = coro 

20 self._obj: _T 

21 

22 def send(self, value) -> None: 

23 return self._coro.send(value) 

24 

25 def throw(self, typ, val=None, tb=None) -> None: 

26 if val is None: 

27 return self._coro.throw(typ) 

28 

29 if tb is None: 

30 return self._coro.throw(typ, val) 

31 

32 return self._coro.throw(typ, val, tb) 

33 

34 def close(self) -> None: 

35 return self._coro.close() 

36 

37 def __await__(self) -> Generator[Any, None, _T]: 

38 return self._coro.__await__() 

39 

40 async def __aenter__(self) -> _T: 

41 self._obj = await self._coro 

42 return self._obj 

43 

44 async def __aexit__(self, exc_type, exc, tb) -> None: 

45 if isinstance(self._obj, Cursor): 

46 await self._obj.close() 

47 

48 

49def contextmanager( 

50 method: Callable[..., Coroutine[Any, Any, _T]] 

51) -> Callable[..., Result[_T]]: 

52 @wraps(method) 

53 def wrapper(self, *args, **kwargs) -> Result[_T]: 

54 return Result(method(self, *args, **kwargs)) 

55 

56 return wrapper