Coverage for netbox/context_managers.py: 100%

1from collections import defaultdict 

2from contextlib import contextmanager 

3 

4from extras.events import flush_events 

5from netbox.context import current_request, events_queue, query_cache 

6from netbox.utils import register_request_processor 

7 

8 

9@register_request_processor 

10@contextmanager 

11def event_tracking(request): 

12 """ 

13 Queue interesting events in memory while processing a request, then flush that queue for processing by the 

14 events pipline before returning the response. 

15 

16 :param request: WSGIRequest object with a unique `id` set 

17 """ 

18 request_token = current_request.set(request) 

19 queue_token = events_queue.set({}) 

20 cache_token = query_cache.set(defaultdict(dict)) 

21 

22 try: 

23 yield 

24 

25 # Flush queued webhooks to RQ. This is done only if the wrapped block completed successfully; events 

26 # queued by a failed request or job must not be dispatched. 

27 if events := list(events_queue.get().values()): 

28 flush_events(events) 

29 

30 finally: 

31 # Restore the previous context vars, whether or not the wrapped block raised an exception 

32 current_request.reset(request_token) 

33 events_queue.reset(queue_token) 

34 query_cache.reset(cache_token)