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 . import imports as __ 

25from . import exceptions as _exceptions 

26from . import generics as _generics 

27 

28 

29async def gather_async( 

30 *operands: __.typx.Any, 

31 return_exceptions: __.typx.Annotated[ 

32 bool, 

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

34 ] = False, 

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

36 ignore_nonawaitables: __.typx.Annotated[ 

37 bool, 

38 __.typx.Doc( 

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

40 ] = False, 

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

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

43 # TODO: Overload signature for 'return_exceptions'. 

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

45 if ignore_nonawaitables: 

46 results = await _gather_async_permissive( operands ) 

47 else: 

48 results = await _gather_async_strict( operands ) 

49 if return_exceptions: return tuple( results ) 

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

51 if errors: raise ExceptionGroup( error_message, errors ) 

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

53 

54 

55async def intercept_error_async( 

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

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

58 ''' Converts unwinding exceptions to error results. 

59 

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

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

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

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

64 behavior. 

65 

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

67 because exceptions can be distinguished from computed values 

68 and collected together into an exception group. 

69 

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

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

72 propagation. 

73 ''' 

74 try: return _generics.Value( await awaitable ) 

75 except Exception as exc: 

76 return _generics.Error( exc ) 

77 

78 

79async def _gather_async_permissive( 

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

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

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

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

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

85 for i, operand in enumerate( operands ): 

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

87 awaitables[ i ] = ( 

88 intercept_error_async( __.typx.cast( 

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

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

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

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

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

94 results[ i ] = result 

95 return results 

96 

97 

98async def _gather_async_strict( 

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

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

101 from inspect import isawaitable, iscoroutine 

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

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

104 for operand in operands: # Sanity check. 

105 if isawaitable( operand ): continue 

106 for operand_ in operands: # Cleanup. 

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

108 raise _exceptions.AsyncAssertionFailure( operand ) 

109 for operand in operands: 

110 awaitables.append( intercept_error_async( __.typx.cast( # noqa: PERF401 

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

112 return await gather( *awaitables )