Coverage for sources/mimeogram/__/exceptions.py: 71%

35 statements  

« prev     ^ index     » next       coverage.py v7.8.2, created at 2025-06-07 04:07 +0000

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' Family of exceptions for package internals. ''' 

22 

23 

24import contextlib as _contextlib 

25import logging as _logging 

26 

27import exceptiongroup as _exceptiongroup 

28 

29from . import imports as __ 

30 

31 

32class Omniexception( BaseException, metaclass = __.ImmutableClass ): 

33 ''' Base for all exceptions raised internally. ''' 

34 # TODO: Class attribute concealment. 

35 # TODO: Instance attribute concealment and immutability. 

36 

37 _attribute_visibility_includes_: __.cabc.Collection[ str ] = ( 

38 frozenset( ( '__cause__', '__context__', ) ) ) 

39 

40 

41class Omnierror( Omniexception, Exception ): 

42 ''' Base for error exceptions raised internally. ''' 

43 

44 

45class AddressLocateFailure( Omnierror, LookupError ): 

46 ''' Failure to locate address. ''' 

47 

48 def __init__( 

49 self, subject: str, address: __.cabc.Sequence[ str ], part: str 

50 ): 

51 super( ).__init__( 

52 f"Could not locate part '{part}' of address '{address}' " 

53 f"in {subject}." ) 

54 

55 

56class AsyncAssertionFailure( Omnierror, AssertionError, TypeError ): 

57 ''' Assertion of awaitability of entity failed. ''' 

58 

59 def __init__( self, entity: __.typx.Any ): 

60 super( ).__init__( f"Entity must be awaitable: {entity!r}" ) 

61 

62 

63class EntryAssertionFailure( Omnierror, AssertionError, KeyError ): 

64 ''' Assertion of entry in dictionary failed. ''' 

65 

66 def __init__( self, subject: str, name: str ): 

67 super( ).__init__( f"Could not find entry '{name}' in {subject}." ) 

68 

69 

70class OperationInvalidity( Omnierror, RuntimeError ): 

71 ''' Invalid operation. ''' 

72 

73 def __init__( self, subject: str, name: str ): 

74 super( ).__init__( 

75 f"Could not perform operation '{name}' on {subject}." ) 

76 

77 

78@_contextlib.contextmanager 

79def report_exceptions( 

80 scribe: _logging.Logger, 

81 message: str, 

82 eclass: type[ BaseException ] = SystemExit, 

83 eposargs: __.cabc.Sequence[ __.typx.Any ] = ( 1, ), 

84) -> __.cabc.Generator[ None, None, None ]: 

85 ''' Intercepts and reports exceptions. 

86 

87 By default, raises ``SystemExit( 1 )``. 

88 ''' 

89 level = scribe.getEffectiveLevel( ) 

90 try: yield 

91 except _exceptiongroup.ExceptionGroup as excg: # pyright: ignore 

92 scribe.error( message ) 

93 for exc in excg.exceptions: # pyright: ignore 

94 if level <= _logging.DEBUG: # noqa: SIM108 

95 nomargs = dict( exc_info = exc ) # pyright: ignore 

96 else: nomargs = { } 

97 scribe.error( 

98 f"\tCause: {exc}", **nomargs ) # pyright: ignore 

99 if eclass: raise eclass( *eposargs ) from None 

100 except Exception as exc: 

101 if level <= _logging.DEBUG: scribe.exception( f"{message}" ) 101 ↛ 103line 101 didn't jump to line 103 because

102 else: scribe.error( f"{message} Cause: {exc}" ) 

103 if eclass: raise eclass( *eposargs ) from None