Coverage for sources/agentsmgr/sources/base.py: 77%

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''' Base abstractions for source handlers. 

22 

23 This module provides the foundational protocols and functions for 

24 resolving various types of data sources to local filesystem paths. 

25''' 

26 

27 

28from .. import nomina as _nomina 

29from . import __ 

30 

31 

32class AbstractSourceHandler( __.immut.Protocol, __.typx.Protocol ): 

33 ''' Protocol for source handlers that resolve specifications to paths. 

34 

35 Source handlers provide a pluggable way to resolve different types 

36 of source specifications (local paths, Git URLs, etc.) to local 

37 filesystem paths where the content can be accessed. 

38 ''' 

39 

40 @__.abc.abstractmethod 

41 def resolve( 

42 self, 

43 source_spec: str, 

44 tag_prefix: _nomina.TagPrefixArgument = __.absent, 

45 ) -> __.Path: 

46 ''' Resolves source specification to local filesystem path. 

47 

48 Returns path to directory containing the resolved source content. 

49 For remote sources, this may involve downloading or cloning to 

50 a temporary location. 

51 ''' 

52 raise NotImplementedError 

53 

54 

55# Private registry mapping URL schemes to source handlers 

56_SCHEME_HANDLERS: __.accret.Dictionary[ str, AbstractSourceHandler ] = ( 

57 __.accret.Dictionary( ) ) 

58 

59_WINDOWS_ABSOLUTE_PATH_MINIMUM_LENGTH = 3 

60 

61 

62def _is_windows_absolute_path( source_spec: str ) -> bool: 

63 ''' Returns true if source specification is a Windows absolute path. ''' 

64 return ( 

65 _WINDOWS_ABSOLUTE_PATH_MINIMUM_LENGTH <= len( source_spec ) 

66 and source_spec[ 1 ] == ':' 

67 and source_spec[ 2 ] in ( '/', '\\' ) 

68 and source_spec[ 0 ].isalpha( ) ) 

69 

70 

71def register_source_handler( 

72 handler: __.typx.Annotated[ 

73 AbstractSourceHandler, 

74 __.ddoc.Doc( ''' The source handler instance ''' ) 

75 ], 

76 schemes: __.typx.Annotated[ 

77 __.cabc.Iterable[ str ], 

78 __.ddoc.Doc( ''' URL schemes this handler supports 

79 (e.g., ['github:', 'gitlab:']) ''' ) 

80 ] 

81) -> None: 

82 ''' Registers a source handler for specific URL schemes. ''' 

83 for scheme in schemes: 

84 _SCHEME_HANDLERS[ scheme ] = handler 

85 

86 

87def source_handler( 

88 schemes: __.typx.Annotated[ 

89 __.cabc.Iterable[ str ], 

90 __.ddoc.Doc( ''' URL schemes this handler supports 

91 (e.g., ['github:', 'gitlab:']) ''' ) 

92 ] 

93) -> __.cabc.Callable[ 

94 [ type[ AbstractSourceHandler ] ], type[ AbstractSourceHandler ] 

95]: 

96 ''' Decorator for automatic source handler registration. 

97 

98 Usage: 

99 @source_handler(['github:', 'gitlab:']) 

100 class GitSourceHandler: 

101 ... 

102 ''' 

103 def decorator( 

104 handler_class: type[ AbstractSourceHandler ] 

105 ) -> type[ AbstractSourceHandler ]: 

106 register_source_handler( handler_class( ), schemes ) 

107 return handler_class 

108 return decorator 

109 

110 

111def resolve_source_location( 

112 source_spec: str, 

113 tag_prefix: _nomina.TagPrefixArgument = __.absent, 

114) -> __.Path: 

115 ''' Resolves data source specification to local filesystem path. 

116 

117 Delegates to registered source handlers based on URL scheme. 

118 Uses urlparse to extract the scheme from the specification. 

119 

120 Raises DataSourceNoSupport if no handler can process the specification. 

121 ''' 

122 if source_spec.startswith( 'git@' ): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 if 'git@' in _SCHEME_HANDLERS: 

124 return _SCHEME_HANDLERS[ 'git@' ].resolve( 

125 source_spec, tag_prefix ) 

126 raise __.DataSourceNoSupport( source_spec ) 

127 if _is_windows_absolute_path( source_spec ): 

128 return _SCHEME_HANDLERS[ '' ].resolve( source_spec, tag_prefix ) 

129 parsed = __.urlparse.urlparse( source_spec ) 

130 if parsed.scheme in _SCHEME_HANDLERS: 130 ↛ 133line 130 didn't jump to line 133 because the condition on line 130 was always true

131 return _SCHEME_HANDLERS[ parsed.scheme ].resolve( 

132 source_spec, tag_prefix ) 

133 raise __.DataSourceNoSupport( source_spec )