Coverage for utilities/exceptions.py: 87%

1from rest_framework import status 

2from rest_framework.exceptions import APIException 

3 

4__all__ = ( 

5 'AbortRequest', 

6 'AbortScript', 

7 'AbortTransaction', 

8 'PermissionsViolation', 

9 'PreconditionFailed', 

10 'RQWorkerNotRunningException', 

11) 

12 

13 

14class AbortTransaction(Exception): 

15 """ 

16 A dummy exception used to trigger a database transaction rollback. 

17 """ 

18 pass 

19 

20 

21class AbortRequest(Exception): 

22 """ 

23 Raised to cleanly abort a request (for example, by a pre_save signal receiver). 

24 """ 

25 def __init__(self, message): 

26 self.message = message 

27 

28 

29class AbortScript(Exception): 

30 """ 

31 Raised to cleanly abort a script. 

32 """ 

33 pass 

34 

35 

36class PermissionsViolation(Exception): 

37 """ 

38 Raised when an operation was prevented because it would violate the 

39 allowed permissions. 

40 """ 

41 message = "Operation failed due to object-level permissions violation" 

42 

43 

44class PreconditionFailed(APIException): 

45 """ 

46 Raised when an If-Match precondition is not satisfied (HTTP 412). 

47 Optionally carries the current ETag so it can be included in the response. 

48 """ 

49 status_code = status.HTTP_412_PRECONDITION_FAILED 

50 default_detail = 'Precondition failed.' 

51 default_code = 'precondition_failed' 

52 

53 def __init__(self, detail=None, code=None, etag=None): 

54 super().__init__(detail=detail, code=code) 

55 self.etag = etag 

56 

57 

58class RQWorkerNotRunningException(APIException): 

59 """ 

60 Indicates the temporary inability to enqueue a new task (e.g. custom script execution) because no RQ worker 

61 processes are currently running. 

62 """ 

63 status_code = status.HTTP_503_SERVICE_UNAVAILABLE 

64 default_detail = 'Unable to process request: RQ worker process not running.' 

65 default_code = 'rq_worker_not_running'