Coverage for utilities/jobs.py: 44%

1from django.contrib import messages 

2from django.utils.safestring import mark_safe 

3from django.utils.translation import gettext_lazy as _ 

4 

5from netbox.jobs import AsyncViewJob 

6from utilities.request import copy_safe_request 

7 

8__all__ = ( 

9 'is_background_request', 

10 'process_request_as_job', 

11) 

12 

13 

14def is_background_request(request): 

15 """ 

16 Return True if the request is being processed as a background job. 

17 """ 

18 return hasattr(request, 'job') 

19 

20 

21def process_request_as_job(view, request, name=None): 

22 """ 

23 Process a request using a view as a background job. 

24 """ 

25 

26 # Check that the request that is not already being processed as a background job (would be a loop) 

27 if is_background_request(request): 

28 return None 

29 

30 # Create a serializable copy of the original request 

31 request_copy = copy_safe_request(request) 

32 

33 # Enqueue a job to perform the work in the background 

34 job = AsyncViewJob.enqueue( 

35 name=name, 

36 user=request.user, 

37 view_cls=view, 

38 request=request_copy, 

39 ) 

40 

41 # Record a message on the original request indicating deferral to a background job 

42 msg = _('Created background job {id}: <a href="{url}">{name}</a>').format( 

43 id=job.pk, 

44 url=job.get_absolute_url(), 

45 name=job.name 

46 ) 

47 messages.info(request, mark_safe(msg)) 

48 

49 return job