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

1import logging 

2from types import TracebackType 

3from typing import Any, Dict, Optional, Type 

4 

5from ._models import Request 

6 

7 

8class Trace: 

9 def __init__( 

10 self, 

11 name: str, 

12 logger: logging.Logger, 

13 request: Optional[Request] = None, 

14 kwargs: Optional[Dict[str, Any]] = None, 

15 ) -> None: 

16 self.name = name 

17 self.logger = logger 

18 self.trace_extension = ( 

19 None if request is None else request.extensions.get("trace") 

20 ) 

21 self.debug = self.logger.isEnabledFor(logging.DEBUG) 

22 self.kwargs = kwargs or {} 

23 self.return_value: Any = None 

24 self.should_trace = self.debug or self.trace_extension is not None 

25 self.prefix = self.logger.name.split(".")[-1] 

26 

27 def trace(self, name: str, info: Dict[str, Any]) -> None: 

28 if self.trace_extension is not None: 

29 prefix_and_name = f"{self.prefix}.{name}" 

30 self.trace_extension(prefix_and_name, info) 

31 

32 if self.debug: 

33 if not info or "return_value" in info and info["return_value"] is None: 

34 message = name 

35 else: 

36 args = " ".join([f"{key}={value!r}" for key, value in info.items()]) 

37 message = f"{name} {args}" 

38 self.logger.debug(message) 

39 

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

41 if self.should_trace: 

42 info = self.kwargs 

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

44 return self 

45 

46 def __exit__( 

47 self, 

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

49 exc_value: Optional[BaseException] = None, 

50 traceback: Optional[TracebackType] = None, 

51 ) -> None: 

52 if self.should_trace: 

53 if exc_value is None: 

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

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

56 else: 

57 info = {"exception": exc_value} 

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

59 

60 async def atrace(self, name: str, info: Dict[str, Any]) -> None: 

61 if self.trace_extension is not None: 

62 prefix_and_name = f"{self.prefix}.{name}" 

63 await self.trace_extension(prefix_and_name, info) 

64 

65 if self.debug: 

66 if not info or "return_value" in info and info["return_value"] is None: 

67 message = name 

68 else: 

69 args = " ".join([f"{key}={value!r}" for key, value in info.items()]) 

70 message = f"{name} {args}" 

71 self.logger.debug(message) 

72 

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

74 if self.should_trace: 

75 info = self.kwargs 

76 await self.atrace(f"{self.name}.started", info) 

77 return self 

78 

79 async def __aexit__( 

80 self, 

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

82 exc_value: Optional[BaseException] = None, 

83 traceback: Optional[TracebackType] = None, 

84 ) -> None: 

85 if self.should_trace: 

86 if exc_value is None: 

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

88 await self.atrace(f"{self.name}.complete", info) 

89 else: 

90 info = {"exception": exc_value} 

91 await self.atrace(f"{self.name}.failed", info)