Coverage for requests_tracker/middleware.py: 100%

1import asyncio 

2import re 

3from typing import Any, Dict 

4from uuid import UUID 

5 

6from django.http import HttpRequest 

7from django.urls import Resolver404, resolve 

8from django.utils.decorators import sync_and_async_middleware 

9 

10from requests_tracker import APP_NAME 

11from requests_tracker.main_request_collector import MainRequestCollector 

12from requests_tracker.settings import debug_application, get_config 

13from requests_tracker.sql.sql_tracker import SQLTracker 

14 

15 

16class RequestWithCollectors(HttpRequest): 

17 request_collectors: Dict[UUID, MainRequestCollector] 

18 

19 

20def is_requests_tracker_request(request: HttpRequest) -> bool: 

21 try: 

22 resolver_match = request.resolver_match or resolve( 

23 request.path, 

24 getattr(request, "urlconf", None), 

25 ) 

26 except Resolver404: 

27 return False 

28 return bool(resolver_match.namespaces and resolver_match.namespaces[-1] == APP_NAME) 

29 

30 

31def is_ignored_request(request: HttpRequest) -> bool: 

32 config = get_config() 

33 

34 if ignore_patterns := config.get("IGNORE_PATHS_PATTERNS"): 

35 for pattern in ignore_patterns: 

36 if re.match(pattern, request.path): 

37 return True 

38 

39 return False 

40 

41 

42async def middleware_async( 

43 request: RequestWithCollectors, 

44 get_response: Any, 

45 request_collectors: Dict[UUID, MainRequestCollector], 

46) -> Any: 

47 if not debug_application(request) or is_ignored_request(request): 

48 return await get_response(request) 

49 

50 if is_requests_tracker_request(request): 

51 if ( 

52 request.method == "DELETE" 

53 and request.path == "/__requests_tracker__/delete" 

54 ): 

55 request_collectors.clear() 

56 request.request_collectors = request_collectors 

57 return await get_response(request) 

58 

59 request_collector = MainRequestCollector(request) 

60 request_collectors[request_collector.request_id] = request_collector 

61 

62 with SQLTracker(request_collector.sql_collector): 

63 response = await get_response(request) 

64 request_collector.wrap_up_request(response) 

65 

66 return response 

67 

68 

69def middleware_sync( 

70 request: RequestWithCollectors, 

71 get_response: Any, 

72 request_collectors: Dict[UUID, MainRequestCollector], 

73) -> Any: 

74 if not debug_application(request) or is_ignored_request(request): 

75 return get_response(request) 

76 

77 if is_requests_tracker_request(request): 

78 if ( 

79 request.method == "DELETE" 

80 and request.path == "/__requests_tracker__/delete" 

81 ): 

82 request_collectors.clear() 

83 request.request_collectors = request_collectors 

84 return get_response(request) 

85 

86 request_collector = MainRequestCollector(request) 

87 request_collectors[request_collector.request_id] = request_collector 

88 

89 with SQLTracker(request_collector.sql_collector): 

90 response = get_response(request) 

91 request_collector.wrap_up_request(response) 

92 

93 return response 

94 

95 

96@sync_and_async_middleware 

97def requests_tracker_middleware( 

98 get_response: Any, 

99) -> Any: 

100 request_collectors: Dict[UUID, MainRequestCollector] = {} 

101 

102 if asyncio.iscoroutinefunction(get_response): 

103 

104 async def middleware(request: RequestWithCollectors) -> Any: 

105 return await middleware_async(request, get_response, request_collectors) 

106 

107 else: 

108 

109 def middleware(request: RequestWithCollectors) -> Any: # type: ignore 

110 return middleware_sync(request, get_response, request_collectors) 

111 

112 return middleware