Coverage for src / lexigram / contracts / exceptions / infra.py: 21%

1"""Infrastructure-related exception classes.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.exceptions.base import LexigramError 

8 

9 

10class InfrastructureError(LexigramError): 

11 """System-level failures (DB down, network timeouts).""" 

12 

13 _code = "LEX_ERR_INFRA_001" 

14 

15 def __init__(self, message: str = "Infrastructure error", **kwargs: Any) -> None: 

16 super().__init__(message, **kwargs) 

17 

18 

19class DatabaseError(InfrastructureError): 

20 """Database-related error (infrastructure-level).""" 

21 

22 _code = "LEX_ERR_DB_001" 

23 

24 def __init__(self, message: str = "Database error", **kwargs: Any) -> None: 

25 super().__init__(message, **kwargs) 

26 

27 

28class MigrationError(InfrastructureError): 

29 """Database migration error.""" 

30 

31 _code = "LEX_ERR_DB_002" 

32 

33 def __init__(self, message: str = "Migration error", **kwargs: Any) -> None: 

34 super().__init__(message, **kwargs) 

35 

36 

37class LockError(InfrastructureError): 

38 """Distributed lock error.""" 

39 

40 _code = "LEX_ERR_INFRA_002" 

41 

42 def __init__(self, message: str = "Lock error", **kwargs: Any) -> None: 

43 super().__init__(message, **kwargs) 

44 

45 

46class LockConflictError(LockError): 

47 """Raised when a lock conflict occurs (resource already locked).""" 

48 

49 _code = "LEX_ERR_INFRA_003" 

50 

51 def __init__( 

52 self, 

53 resource: str, 

54 owner: str | None = None, 

55 message: str | None = None, 

56 **kwargs: Any, 

57 ) -> None: 

58 if message is None: 

59 message = f"Lock conflict on '{resource}'" 

60 if owner: 

61 message += f" (held by: {owner})" 

62 

63 super().__init__( 

64 message, 

65 details={"resource": resource, "owner": owner}, 

66 **kwargs, 

67 ) 

68 self.resource = resource 

69 self.owner = owner 

70 

71 

72class RegistryError(LexigramError): 

73 """Base exception for registry errors.""" 

74 

75 _code = "LEX_ERR_REG_001" 

76 

77 def __init__(self, message: str = "Registry error", **kwargs: Any) -> None: 

78 super().__init__(message, **kwargs) 

79 

80 

81class RegistryKeyError(RegistryError, KeyError): 

82 """Raised when a key is not found in a registry.""" 

83 

84 _code = "LEX_ERR_REG_002" 

85 

86 def __init__(self, message: str = "Key not found", **kwargs: Any) -> None: 

87 super().__init__(message, **kwargs) 

88 

89 

90class RegistryAlreadyExistsError(RegistryError): 

91 """Raised when attempting to register a duplicate key in a registry.""" 

92 

93 _code = "LEX_ERR_REG_003" 

94 

95 def __init__(self, message: str = "Key already exists", **kwargs: Any) -> None: 

96 super().__init__(message, **kwargs) 

97 

98 

99class IntegrityError(DatabaseError): 

100 """Raised when a database integrity constraint is violated.""" 

101 

102 _code = "LEX_ERR_DB_003" 

103 

104 def __init__( 

105 self, message: str = "Integrity constraint violated", **kwargs: Any 

106 ) -> None: 

107 super().__init__(message, **kwargs) 

108 

109 

110class ConstraintError(IntegrityError): 

111 """Raised when a generic database constraint is violated.""" 

112 

113 _code = "LEX_ERR_DB_004" 

114 

115 def __init__(self, message: str = "Constraint violated", **kwargs: Any) -> None: 

116 super().__init__(message, **kwargs) 

117 

118 

119class DuplicateKeyError(ConstraintError): 

120 """Raised when a unique-key constraint is violated (duplicate record).""" 

121 

122 _code = "LEX_ERR_DB_005" 

123 

124 def __init__(self, message: str = "Duplicate key", **kwargs: Any) -> None: 

125 super().__init__(message, **kwargs) 

126 

127 

128class NoPrimaryBackendError(DatabaseError): 

129 """Raised when a multi-backend setup has no backend marked as primary. 

130 

131 For multi-backend SQL configurations, exactly one backend must be marked 

132 with primary: true. This error is raised when no backend is marked primary. 

133 """ 

134 

135 _code = "LEX_ERR_DB_011" 

136 

137 

138__all__ = [ 

139 "ConstraintError", 

140 "DatabaseError", 

141 "DuplicateKeyError", 

142 "InfrastructureError", 

143 "IntegrityError", 

144 "LockConflictError", 

145 "LockError", 

146 "MigrationError", 

147 "NoPrimaryBackendError", 

148 "RegistryAlreadyExistsError", 

149 "RegistryError", 

150 "RegistryKeyError", 

151]