Coverage for sources/librovore/xtnsmgr/processors.py: 47%

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''' Processor loading and registration management. ''' 

22 

23 

24from . import __ 

25from . import cachemgr as _cachemgr 

26from . import configuration as _configuration 

27from . import importation as _importation 

28 

29 

30_scribe = __.acquire_scribe( __name__ ) 

31 

32 

33def _raise_registration_error( name: str ) -> None: 

34 ''' Raises registration error for missing function. ''' 

35 raise __.ExtensionConfigurationInvalidity( 

36 name, "No registration function found" ) 

37 

38 

39async def register_processors( auxdata: __.Globals ): 

40 ''' Registers inventory and structure processors from configuration. ''' 

41 await _register_processor_type( 

42 auxdata, 'inventory', _configuration.extract_inventory_extensions ) 

43 await _register_processor_type( 

44 auxdata, 'structure', _configuration.extract_structure_extensions ) 

45 

46 

47async def _register_processor_type( 

48 auxdata: __.Globals, 

49 processor_type: str, 

50 extract_func: __.cabc.Callable[ [ __.Globals ], __.typx.Any ] 

51) -> None: 

52 ''' Registers processors of specific type based on configuration. ''' 

53 try: extensions = extract_func( auxdata ) 

54 except ( KeyError, ValueError, TypeError ) as exc: 

55 _scribe.error( f"{processor_type.title()} configuration loading " 

56 f"failed: {exc}." ) 

57 return 

58 active_extensions = _configuration.select_active_extensions( extensions ) 

59 if not active_extensions: return 

60 intrinsic_extensions = ( 

61 _configuration.select_intrinsic_extensions( active_extensions ) ) 

62 external_extensions = tuple( 

63 ext for ext in active_extensions 

64 if ext.get( 'package' ) and ext not in intrinsic_extensions ) 

65 await _ensure_external_packages( external_extensions ) 

66 if not intrinsic_extensions and not external_extensions: 

67 _scribe.warning( f"No {processor_type} processors could be loaded." ) 

68 return 

69 for extension in active_extensions: 

70 _register_extension( extension, processor_type ) 

71 

72 

73async def _ensure_external_packages( 

74 extensions: __.cabc.Sequence[ _configuration.ExtensionConfig ] 

75) -> None: 

76 ''' Ensures external packages are installed and importable in parallel. ''' 

77 if not extensions: return 

78 specifications = [ ext[ 'package' ] for ext in extensions ] 

79 count = len( specifications ) 

80 _scribe.info( f"Ensuring {count} external packages available." ) 

81 tasks = [ _cachemgr.ensure_package( spec ) for spec in specifications ] 

82 await __.asyncf.gather_async( 

83 *tasks, error_message = "Failed to install external packages." ) 

84 

85 

86def _register_extension( 

87 extension: _configuration.ExtensionConfig, 

88 processor_type: str 

89) -> None: 

90 ''' Registers extension from configuration to appropriate registry. ''' 

91 name = extension[ 'name' ] 

92 arguments = _configuration.extract_extension_arguments( extension ) 

93 if 'package' not in extension: 

94 if processor_type == 'inventory': 94 ↛ 97line 94 didn't jump to line 97 because the condition on line 94 was always true

95 module_name = f"{__.package_name}.inventories.{name}" 

96 else: 

97 module_name = f"{__.package_name}.structures.{name}" 

98 else: module_name = name 

99 try: module = _importation.import_processor_module( module_name ) 

100 except ( ImportError, ModuleNotFoundError ) as exc: 

101 _scribe.error( f"Failed to import {processor_type} processor " 

102 f"{name}: {exc}" ) 

103 return 

104 try: 

105 if hasattr( module, 'register' ): 

106 module.register( arguments ) 

107 else: 

108 _raise_registration_error( name ) 

109 except Exception as exc: 

110 _scribe.error( f"Failed to register {processor_type} processor " 

111 f"{name}: {exc}" ) 

112 return 

113 _scribe.info( f"Registered {processor_type} extension: {name}." )