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

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 Policy 

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, role: str 

34) -> Never: 

35 match err: 

36 case IntegrityError( 

37 orig=Sqlite3IntegrityError(sqlite_errorname=str(msg)) 

38 | PsycopgUniqueViolation(diag=PsycopgDiagnostic(message_detail=str(msg))) 

39 ) if msg == "SQLITE_CONSTRAINT_PRIMARYKEY" or ( 

40 f"Key (role_id, subject_id)=({role}, {subject}) already exists." in msg 

41 ): 

42 raise PyPermissionError( 

43 f"Conflict: Role with ID '{role}' already assigned to Subject with ID '{subject}'!" 

44 ) 

45 case IntegrityError( 

46 orig=Sqlite3IntegrityError(sqlite_errorname="SQLITE_CONSTRAINT_FOREIGNKEY") 

47 ): 

48 raise PyPermissionError( 

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

50 ) 

51 case IntegrityError( 

52 orig=PsycopgForeignKeyViolation( 

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

54 ) 

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

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

57 case IntegrityError( 

58 orig=PsycopgForeignKeyViolation( 

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

60 ) 

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

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

63 case _: # pragma: no cover 

64 ... 

65 raise PyPermissionError("Unexpected IntegrityError") # pragma: no cover 

66 

67 

68def process_policy_integrity_error( 

69 *, 

70 err: IntegrityError, 

71 policy: Policy, 

72) -> Never: 

73 match err: 

74 case IntegrityError( 

75 orig=Sqlite3IntegrityError(sqlite_errorname="SQLITE_CONSTRAINT_PRIMARYKEY") 

76 | PsycopgUniqueViolation(diag=PsycopgDiagnostic(message_detail=str())) 

77 ): 

78 raise PyPermissionError( 

79 f"Conflict: Policy '{str(policy)}' already exists!" 

80 ) from err 

81 case IntegrityError( 

82 orig=Sqlite3IntegrityError(sqlite_errorname="SQLITE_CONSTRAINT_FOREIGNKEY") 

83 | PsycopgForeignKeyViolation(diag=PsycopgDiagnostic(message_detail=str())) 

84 ): 

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

86 case _: # pragma: no cover 

87 ... 

88 raise PyPermissionError("Unexpected IntegrityError") # pragma: no cover