Coverage for netrun_errors \ service.py: 100%

1""" 

2Service-related exceptions for Netrun Systems. 

3 

4Exceptions for service availability, external dependencies, 

5and system-level errors. 

6""" 

7 

8from typing import Any, Dict, Optional 

9 

10from fastapi import status 

11 

12from .base import NetrunException 

13 

14 

15class ServiceUnavailableError(NetrunException): 

16 """ 

17 Raised when service or dependency is temporarily unavailable. 

18 

19 Common scenarios: 

20 - Database connection failures 

21 - External API timeouts 

22 - Maintenance mode 

23 - Circuit breaker open 

24 

25 Status Code: 503 Service Unavailable 

26 Error Code: SERVICE_UNAVAILABLE 

27 """ 

28 

29 def __init__( 

30 self, 

31 message: str = "Service is temporarily unavailable. Please try again later.", 

32 details: Optional[Dict[str, Any]] = None, 

33 correlation_id: Optional[str] = None, 

34 ): 

35 super().__init__( 

36 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

37 error_code="SERVICE_UNAVAILABLE", 

38 message=message, 

39 details=details, 

40 correlation_id=correlation_id, 

41 ) 

42 

43 

44class TemporalUnavailableError(NetrunException): 

45 """ 

46 Raised when Temporal workflow engine is unavailable. 

47 

48 Specific to Netrun Systems' Temporal.io integration for 

49 long-running workflows and distributed transactions. 

50 

51 Status Code: 503 Service Unavailable 

52 Error Code: TEMPORAL_UNAVAILABLE 

53 """ 

54 

55 def __init__( 

56 self, 

57 message: str = "Workflow engine is temporarily unavailable. Please try again later.", 

58 details: Optional[Dict[str, Any]] = None, 

59 correlation_id: Optional[str] = None, 

60 ): 

61 super().__init__( 

62 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

63 error_code="TEMPORAL_UNAVAILABLE", 

64 message=message, 

65 details=details, 

66 correlation_id=correlation_id, 

67 )