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

1import inspect 

2import logging 

3from types import TracebackType 

4from typing import Any, Dict, Optional, Type 

5 

6from ._models import Request 

7 

8 

9class Trace: 

10 def __init__( 

11 self, 

12 name: str, 

13 logger: logging.Logger, 

14 request: Optional[Request] = None, 

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

16 ) -> None: 

17 self.name = name 

18 self.logger = logger 

19 self.trace_extension = ( 

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

21 ) 

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

23 self.kwargs = kwargs or {} 

24 self.return_value: Any = None 

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

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

27 

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

29 if self.trace_extension is not None: 

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

31 ret = self.trace_extension(prefix_and_name, info) 

32 if inspect.iscoroutine(ret): # pragma: no cover 

33 raise TypeError( 

34 "If you are using a synchronous interface, " 

35 "the callback of the `trace` extension should " 

36 "be a normal function instead of an asynchronous function." 

37 ) 

38 

39 if self.debug: 

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

41 message = name 

42 else: 

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

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

45 self.logger.debug(message) 

46 

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

48 if self.should_trace: 

49 info = self.kwargs 

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

51 return self 

52 

53 def __exit__( 

54 self, 

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

56 exc_value: Optional[BaseException] = None, 

57 traceback: Optional[TracebackType] = None, 

58 ) -> None: 

59 if self.should_trace: 

60 if exc_value is None: 

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

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

63 else: 

64 info = {"exception": exc_value} 

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

66 

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

68 if self.trace_extension is not None: 

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

70 coro = self.trace_extension(prefix_and_name, info) 

71 if not inspect.iscoroutine(coro): # pragma: no cover 

72 raise TypeError( 

73 "If you're using an asynchronous interface, " 

74 "the callback of the `trace` extension should " 

75 "be an asynchronous function rather than a normal function." 

76 ) 

77 await coro 

78 

79 if self.debug: 

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

81 message = name 

82 else: 

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

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

85 self.logger.debug(message) 

86 

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

88 if self.should_trace: 

89 info = self.kwargs 

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

91 return self 

92 

93 async def __aexit__( 

94 self, 

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

96 exc_value: Optional[BaseException] = None, 

97 traceback: Optional[TracebackType] = None, 

98 ) -> None: 

99 if self.should_trace: 

100 if exc_value is None: 

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

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

103 else: 

104 info = {"exception": exc_value} 

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