Coverage for src/pypermission/util/exception_handling.py: 71%

1from sqlite3 import IntegrityError as Sqlite3IntegrityError 

2from typing import TYPE_CHECKING, Never 

3 

4from sqlalchemy.exc import IntegrityError 

5 

6from pypermission.exc import PyPermissionError 

7from pypermission.models import Permission 

8 

9if TYPE_CHECKING: 

10 from psycopg.errors import Diagnostic as PsycopgDiagnostic 

11 from psycopg.errors import ForeignKeyViolation as PsycopgForeignKeyViolation 

12 from psycopg.errors import UniqueViolation as PsycopgUniqueViolation 

13else: 

14 try: 

15 from psycopg.errors import Diagnostic as PsycopgDiagnostic 

16 from psycopg.errors import ForeignKeyViolation as PsycopgForeignKeyViolation 

17 from psycopg.errors import UniqueViolation as PsycopgUniqueViolation 

18 except ModuleNotFoundError: 

19 

20 class PsycopgForeignKeyViolation(Exception): ... 

21 

22 class PsycopgUniqueViolation(Exception): ... 

23 

24 class PsycopgDiagnostic: ... 

25 

26 

27################################################################################ 

28#### Error Utilities 

29################################################################################ 

30 

31 

32def process_subject_role_integrity_error( 

33 *, err: IntegrityError, subject: str | None = None, role: str | None = None 

34) -> Never: 

35 match err: 

36 case IntegrityError(orig=Sqlite3IntegrityError()): 

37 if subject is not None and role is not None: 

38 raise PyPermissionError( 

39 f"Subject '{subject}' or Role '{role}' does not exist!" 

40 ) 

41 if subject is not None: 

42 raise PyPermissionError(f"Subject '{subject}' does not exist!") 

43 if role is not None: 

44 raise PyPermissionError(f"Role '{role}' does not exist!") 

45 case IntegrityError( 

46 orig=PsycopgForeignKeyViolation( 

47 diag=PsycopgDiagnostic(message_detail=str(msg)) 

48 ) 

49 ): 

50 if f"Key (role_id)=({role}) is not present in table" in msg: 

51 raise PyPermissionError(f"Role '{role}' does not exist!") 

52 if f"Key (subject_id)=({subject}) is not present in table" in msg: 

53 raise PyPermissionError(f"Subject '{subject}' does not exist!") 

54 case _: 

55 ... 

56 raise PyPermissionError("Unexpected IntegrityError") 

57 

58 

59def process_policy_integrity_error( 

60 *, 

61 err: IntegrityError, 

62 role: str, 

63 permission: Permission, 

64) -> Never: 

65 match err: 

66 case IntegrityError( 

67 orig=Sqlite3IntegrityError(sqlite_errorname="SQLITE_CONSTRAINT_PRIMARYKEY") 

68 ): 

69 raise PyPermissionError( 

70 f"Permission '{str(permission)}' does already exist!" 

71 ) from err 

72 case IntegrityError( 

73 orig=Sqlite3IntegrityError(sqlite_errorname="SQLITE_CONSTRAINT_FOREIGNKEY") 

74 ): 

75 raise PyPermissionError(f"Role '{role}' does not exist!") from err 

76 case IntegrityError( 

77 orig=PsycopgUniqueViolation( 

78 diag=PsycopgDiagnostic(message_detail=str(_msg)) 

79 ) 

80 ): 

81 raise PyPermissionError( 

82 f"Permission '{str(permission)}' does already exist!" 

83 ) from err 

84 case IntegrityError( 

85 orig=PsycopgForeignKeyViolation( 

86 diag=PsycopgDiagnostic(message_detail=str(_msg)) 

87 ) 

88 ): 

89 raise PyPermissionError(f"Role '{role}' does not exist!") from err 

90 case _: 

91 ... 

92 raise PyPermissionError("Unexpected IntegrityError")