Coverage for dataexcept/database_exceptions.py: 100%

1"""Custom exceptions for database operations.""" 

2 

3from __future__ import annotations 

4 

5from ._causes import resolve_cause 

6from .base import DataExceptError 

7from .redaction import redact_url 

8 

9 

10class DatabaseError(DataExceptError): 

11 """Base exception for database-related errors.""" 

12 

13 pass 

14 

15 

16class DatabaseConnectionError(DatabaseError): 

17 """Raised when connecting to the database fails.""" 

18 

19 def __init__(self, db_url: str, message: str | None = None) -> None: 

20 """Initialize DatabaseConnectionError. 

21 

22 Args: 

23 db_url: Database connection URL. 

24 message: Optional custom error message. 

25 """ 

26 # A connection URL routinely carries a username and password. 

27 self.db_url = redact_url(db_url) 

28 default = f"Failed to connect to database at '{self.db_url}'" 

29 super().__init__(message or default) 

30 

31 

32class QueryExecutionError(DatabaseError): 

33 """Raised when a database query execution fails.""" 

34 

35 def __init__( 

36 self, 

37 query: str, 

38 original: Exception | None = None, 

39 *, 

40 cause: Exception | None = None, 

41 ) -> None: 

42 """Initialize QueryExecutionError. 

43 

44 Args: 

45 query: SQL query string. 

46 original: Legacy alias for ``cause``. 

47 cause: Optional underlying exception. 

48 """ 

49 resolved = resolve_cause(cause=cause, original=original) 

50 self.query = query 

51 self.original = resolved 

52 self.cause = resolved 

53 msg = f"Query failed: {query}" 

54 if resolved: 

55 msg += f" ({resolved})" 

56 super().__init__(msg) 

57 

58 

59class TransactionError(DatabaseError): 

60 """Raised when a database transaction fails.""" 

61 

62 def __init__( 

63 self, 

64 transaction_id: str | None = None, 

65 message: str | None = None, 

66 *, 

67 cause: Exception | None = None, 

68 ) -> None: 

69 """Initialize TransactionError. 

70 

71 Args: 

72 transaction_id: Identifier for the transaction. 

73 message: Optional custom error message. 

74 cause: Optional underlying exception. 

75 """ 

76 self.transaction_id = transaction_id 

77 self.cause = cause 

78 default = "Database transaction failed" 

79 if transaction_id: 

80 default += f" (id={transaction_id})" 

81 super().__init__(message or default) 

82 

83 

84__all__ = [ 

85 "DatabaseError", 

86 "DatabaseConnectionError", 

87 "QueryExecutionError", 

88 "TransactionError", 

89]