Coverage for sources/librovore/inventories/sphinx/detection.py: 19%

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''' Inventory detection implementations. ''' 

22 

23 

24from urllib.parse import ParseResult as _Url 

25 

26import sphobjinv as _sphobjinv 

27 

28from . import __ 

29 

30 

31class SphinxInventoryDetection( __.InventoryDetection ): 

32 ''' Detection result for Sphinx inventory sources. ''' 

33 

34 @classmethod 

35 async def from_source( 

36 selfclass, 

37 auxdata: __.ApplicationGlobals, 

38 processor: __.Processor, 

39 source: str, 

40 ) -> __.typx.Self: 

41 ''' Constructs Sphinx inventory detection from source. ''' 

42 # TODO: Figure out why this is not used. 

43 # This is not used in current implementation 

44 return selfclass( processor = processor, confidence = 0.0 ) 

45 

46 async def filter_inventory( 

47 self, 

48 auxdata: __.ApplicationGlobals, 

49 source: str, /, *, 

50 filters: __.cabc.Mapping[ str, __.typx.Any ], 

51 details: __.InventoryQueryDetails = ( 

52 __.InventoryQueryDetails.Documentation ), 

53 ) -> list[ dict[ str, __.typx.Any ] ]: 

54 ''' Filters inventory objects from Sphinx source. ''' 

55 return await filter_inventory( 

56 source, filters = filters, details = details ) 

57 

58 

59def derive_inventory_url( base_url: _Url ) -> _Url: 

60 ''' Derives objects.inv URL from base URL ParseResult. ''' 

61 new_path = f"{base_url.path}/objects.inv" 

62 # TODO: Do not rely on named tuple internals. 

63 return base_url._replace( path = new_path ) 

64 

65 

66def extract_inventory( base_url: _Url ) -> _sphobjinv.Inventory: 

67 ''' Extracts and parses Sphinx inventory from URL or file path. ''' 

68 url = derive_inventory_url( base_url ) 

69 url_s = url.geturl( ) 

70 nomargs: __.NominativeArguments = { } 

71 match url.scheme: 

72 case 'http' | 'https': nomargs[ 'url' ] = url_s 

73 case 'file': nomargs[ 'fname_zlib' ] = url.path 

74 case _: 

75 raise __.InventoryUrlNoSupport( 

76 url, component = 'scheme', value = url.scheme ) 

77 try: return _sphobjinv.Inventory( **nomargs ) 

78 except ( ConnectionError, OSError, TimeoutError ) as exc: 

79 raise __.InventoryInaccessibility( 

80 url_s, cause = exc ) from exc 

81 except Exception as exc: 

82 raise __.InventoryInvalidity( url_s, cause = exc ) from exc 

83 

84 

85async def filter_inventory( 

86 source: str, /, *, 

87 filters: __.cabc.Mapping[ str, __.typx.Any ], 

88 details: __.InventoryQueryDetails = ( 

89 __.InventoryQueryDetails.Documentation ), 

90) -> list[ dict[ str, __.typx.Any ] ]: 

91 ''' Extracts and filters inventory objects by structural criteria only. ''' 

92 domain = filters.get( 'domain', '' ) or __.absent 

93 role = filters.get( 'role', '' ) or __.absent 

94 priority = filters.get( 'priority', '' ) or __.absent 

95 base_url = __.normalize_base_url( source ) 

96 inventory = extract_inventory( base_url ) 

97 all_objects: list[ dict[ str, __.typx.Any ] ] = [ ] 

98 for objct in inventory.objects: 

99 if not __.is_absent( domain ) and objct.domain != domain: continue 

100 if not __.is_absent( role ) and objct.role != role: continue 

101 if not __.is_absent( priority ) and objct.priority != priority: 

102 continue 

103 obj = dict( format_inventory_object( objct ) ) 

104 obj[ '_inventory_project' ] = inventory.project 

105 obj[ '_inventory_version' ] = inventory.version 

106 all_objects.append( obj ) 

107 return all_objects 

108 

109 

110def format_inventory_object( 

111 objct: __.typx.Any, 

112) -> __.cabc.Mapping[ str, __.typx.Any ]: 

113 ''' Formats an inventory object for output. ''' 

114 return { 

115 'name': objct.name, 

116 'domain': objct.domain, 

117 'role': objct.role, 

118 'priority': objct.priority, 

119 'uri': objct.uri, 

120 'dispname': ( 

121 objct.dispname if objct.dispname != '-' else objct.name 

122 ), 

123 }