Coverage for dataexcept/exceptions/external.py: 100%

1from .._causes import resolve_cause 

2from .base import JobError 

3 

4 

5class ServiceConnectionError(JobError): 

6 """Raised when a connection to an external service fails.""" 

7 

8 def __init__( 

9 self, 

10 service_name: str, 

11 original_exception: Exception | None = None, 

12 *, 

13 cause: Exception | None = None, 

14 ): 

15 resolved = resolve_cause( 

16 cause=cause, 

17 original_exception=original_exception, 

18 ) 

19 self.service_name = service_name 

20 self.original_exception = resolved 

21 self.cause = resolved 

22 msg = f"Failed to connect to service '{service_name}'" 

23 if resolved: 

24 msg += f": {resolved}" 

25 super().__init__(msg) 

26 

27 

28class OperationTimeoutError(JobError): 

29 """Raised when an operation exceeds its time limit.""" 

30 

31 def __init__( 

32 self, 

33 operation: str, 

34 timeout: float, 

35 *, 

36 cause: Exception | None = None, 

37 ): 

38 self.operation = operation 

39 self.timeout = timeout 

40 self.cause = cause 

41 msg = f"Operation '{operation}' timed out after {timeout} seconds" 

42 super().__init__(msg) 

43 

44 

45class ResourceNotFoundError(JobError): 

46 """Raised when a required resource cannot be found.""" 

47 

48 def __init__(self, resource_type: str, identifier: str): 

49 self.resource_type = resource_type 

50 self.identifier = identifier 

51 msg = f"{resource_type} with identifier '{identifier}' not found" 

52 super().__init__(msg) 

53 

54 

55class DependencyError(JobError): 

56 """Raised when a job dependency is missing or fails.""" 

57 

58 def __init__(self, dependency_name: str, message: str | None = None): 

59 self.dependency_name = dependency_name 

60 self.message = message or f"Dependency '{dependency_name}' error" 

61 super().__init__(self.message)