Coverage for requests_tracker/main_request_collector.py: 100%

1from datetime import datetime 

2from typing import Any, Dict, Optional 

3from uuid import UUID, uuid4 

4 

5from django.http import HttpRequest, HttpResponse 

6from django.urls import Resolver404, resolve 

7 

8from requests_tracker.base_collector import Collector 

9from requests_tracker.headers.header_collector import HeaderCollector 

10from requests_tracker.sql.sql_collector import SQLCollector 

11 

12 

13class MainRequestCollector: 

14 request_id: UUID 

15 request: HttpRequest 

16 django_view: str 

17 start_time: datetime 

18 end_time: Optional[datetime] 

19 response: Optional[HttpResponse] 

20 

21 sql_collector: SQLCollector 

22 header_collector: HeaderCollector 

23 

24 def __init__(self, request: HttpRequest): 

25 self.request_id = uuid4() 

26 self.request = request 

27 try: 

28 self.django_view = resolve(self.request.path)._func_path 

29 except Resolver404: 

30 self.django_view = "NOT FOUND" 

31 self.start_time = datetime.now() 

32 self.end_time = None 

33 self.response = None 

34 

35 self.sql_collector = SQLCollector() 

36 self.header_collector = HeaderCollector() 

37 

38 def wrap_up_request(self, response: HttpResponse) -> None: 

39 """ 

40 Called after Django has processed the request, before response is returned 

41 """ 

42 self.set_end_time() 

43 self.response = response 

44 self.header_collector.process_request(self.request, self.response) 

45 

46 @property 

47 def duration(self) -> Optional[int]: 

48 """duration in milliseconds""" 

49 return ( 

50 int((self.end_time - self.start_time).total_seconds() * 1000) 

51 if self.end_time is not None 

52 else None 

53 ) 

54 

55 @property 

56 def finished(self) -> bool: 

57 return self.response is not None 

58 

59 def set_end_time(self) -> None: 

60 self.end_time = datetime.now() 

61 

62 def get_collectors(self) -> Dict[str, Collector]: 

63 collectors: Dict[str, Collector] = {} 

64 

65 for attribute_name, attribute_value in self.__dict__.items(): 

66 if isinstance(attribute_value, Collector): 

67 attribute_value.generate_statistics() 

68 collectors[attribute_name] = attribute_value 

69 

70 return collectors 

71 

72 def get_as_context(self) -> Dict[str, Any]: 

73 return { 

74 "request": self.request, 

75 "request_id": self.request_id, 

76 "django_view": self.django_view, 

77 "start_time": self.start_time, 

78 "end_time": self.end_time, 

79 "duration": self.duration, 

80 "response": self.response, 

81 "finished": self.finished, 

82 **self.get_collectors(), 

83 } 

84 

85 def matches_search_filter(self, search: str) -> bool: 

86 search = search.lower() 

87 return ( 

88 search in self.request.path.lower() 

89 or search in self.django_view.lower() 

90 or next( 

91 ( 

92 True 

93 for collector in self.get_collectors().values() 

94 if collector.matches_search_filter(search) 

95 ), 

96 False, 

97 ) 

98 )