Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpcore/_trace.py: 27%

1from types import TracebackType 

2from typing import Any, Dict, Optional, Type 

3 

4from ._models import Request 

5 

6 

7class Trace: 

8 def __init__( 

9 self, name: str, request: Request, kwargs: Optional[Dict[str, Any]] = None 

10 ) -> None: 

11 self.name = name 

12 self.trace = request.extensions.get("trace") 

13 self.kwargs = kwargs or {} 

14 self.return_value: Any = None 

15 

16 def __enter__(self) -> "Trace": 

17 if self.trace is not None: 

18 info = self.kwargs 

19 self.trace(f"{self.name}.started", info) 

20 return self 

21 

22 def __exit__( 

23 self, 

24 exc_type: Optional[Type[BaseException]] = None, 

25 exc_value: Optional[BaseException] = None, 

26 traceback: Optional[TracebackType] = None, 

27 ) -> None: 

28 if self.trace is not None: 

29 if exc_value is None: 

30 info = {"return_value": self.return_value} 

31 self.trace(f"{self.name}.complete", info) 

32 else: 

33 info = {"exception": exc_value} 

34 self.trace(f"{self.name}.failed", info) 

35 

36 async def __aenter__(self) -> "Trace": 

37 if self.trace is not None: 

38 info = self.kwargs 

39 await self.trace(f"{self.name}.started", info) 

40 return self 

41 

42 async def __aexit__( 

43 self, 

44 exc_type: Optional[Type[BaseException]] = None, 

45 exc_value: Optional[BaseException] = None, 

46 traceback: Optional[TracebackType] = None, 

47 ) -> None: 

48 if self.trace is not None: 

49 if exc_value is None: 

50 info = {"return_value": self.return_value} 

51 await self.trace(f"{self.name}.complete", info) 

52 else: 

53 info = {"exception": exc_value} 

54 await self.trace(f"{self.name}.failed", info)