Coverage for sources/mimeogram/__/asyncf.py: 100%

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''' Helper functions for async execution. ''' 

22 

23 

24from __future__ import annotations 

25 

26from . import imports as __ 

27from . import exceptions as _exceptions 

28from . import generics as _generics 

29 

30 

31async def gather_async( 

32 *operands: __.typx.Any, 

33 return_exceptions: __.typx.Annotated[ 

34 bool, 

35 __.typx.Doc( ''' Raw or wrapped results. Wrapped, if true. ''' ) 

36 ] = False, 

37 error_message: str = 'Failure of async operations.', 

38 ignore_nonawaitables: __.typx.Annotated[ 

39 bool, 

40 __.typx.Doc( 

41 ''' Ignore or error on non-awaitables. Ignore, if true. ''' ) 

42 ] = False, 

43) -> tuple[ __.typx.Any, ... ]: 

44 ''' Gathers results from invocables concurrently and asynchronously. ''' 

45 from exceptiongroup import ExceptionGroup # TODO: Python 3.11: builtin 

46 if ignore_nonawaitables: 

47 results = await _gather_async_permissive( operands ) 

48 else: 

49 results = await _gather_async_strict( operands ) 

50 if return_exceptions: return tuple( results ) 

51 errors = tuple( result.error for result in results if result.is_error( ) ) 

52 if errors: raise ExceptionGroup( error_message, errors ) 

53 return tuple( result.extract( ) for result in results ) 

54 

55 

56async def intercept_error_async( 

57 awaitable: __.cabc.Awaitable[ __.typx.Any ] 

58) -> _generics.Result[ object, Exception ]: 

59 ''' Converts unwinding exceptions to error results. 

60 

61 Exceptions, which are not instances of :py:exc:`Exception` or one of 

62 its subclasses, are allowed to propagate. In particular, 

63 :py:exc:`KeyboardInterrupt` and :py:exc:`SystemExit` must be allowed 

64 to propagate to be consistent with :py:class:`asyncio.TaskGroup` 

65 behavior. 

66 

67 Helpful when working with :py:func:`asyncio.gather`, for example, 

68 because exceptions can be distinguished from computed values 

69 and collected together into an exception group. 

70 

71 In general, it is a bad idea to swallow exceptions. In this case, 

72 the intent is to add them into an exception group for continued 

73 propagation. 

74 ''' 

75 try: return _generics.Value( await awaitable ) 

76 except Exception as exc: # pylint: disable=broad-exception-caught 

77 return _generics.Error( exc ) 

78 

79 

80async def _gather_async_permissive( 

81 operands: __.cabc.Sequence[ __.typx.Any ] 

82) -> __.cabc.Sequence[ __.typx.Any ]: 

83 from asyncio import gather # TODO? Python 3.11: TaskGroup 

84 awaitables: dict[ int, __.cabc.Awaitable[ __.typx.Any ] ] = { } 

85 results: list[ _generics.GenericResult ] = [ ] 

86 for i, operand in enumerate( operands ): 

87 if isinstance( operand, __.cabc.Awaitable ): 

88 awaitables[ i ] = ( 

89 intercept_error_async( __.typx.cast( 

90 __.cabc.Awaitable[ __.typx.Any ], operand ) ) ) 

91 results.append( _generics.Value( None ) ) 

92 else: results.append( _generics.Value( operand ) ) 

93 results_ = await gather( *awaitables.values( ) ) 

94 for i, result in zip( awaitables.keys( ), results_ ): 

95 results[ i ] = result 

96 return results 

97 

98 

99async def _gather_async_strict( 

100 operands: __.cabc.Sequence[ __.typx.Any ] 

101) -> __.cabc.Sequence[ __.typx.Any ]: 

102 from inspect import isawaitable, iscoroutine 

103 from asyncio import gather # TODO? Python 3.11: TaskGroup 

104 awaitables: list[ __.cabc.Awaitable[ __.typx.Any ] ] = [ ] 

105 for operand in operands: # Sanity check. 

106 if isawaitable( operand ): continue 

107 for operand_ in operands: # Cleanup. 

108 if iscoroutine( operand_ ): operand_.close( ) 

109 raise _exceptions.AsyncAssertionFailure( operand ) 

110 for operand in operands: 

111 awaitables.append( intercept_error_async( __.typx.cast( 

112 __.cabc.Awaitable[ __.typx.Any ], operand ) ) ) 

113 return await gather( *awaitables )