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

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 __ 

29 

30 

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

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

33 

34 Source handlers provide a pluggable way to resolve different types 

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

36 filesystem paths where the content can be accessed. 

37 ''' 

38 

39 @__.abc.abstractmethod 

40 def resolve( 

41 self, 

42 source_spec: str, 

43 tag_prefix: __.typx.Annotated[ 

44 __.Absential[ str ], 

45 __.ddoc.Doc( 

46 "Prefix for filtering version tags when no explicit ref " 

47 "is specified. Only tags starting with this prefix will be " 

48 "considered, and the prefix will be stripped before version " 

49 "parsing." ), 

50 ] = __.absent, 

51 ) -> __.Path: 

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

53 

54 Returns path to directory containing the resolved source content. 

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

56 a temporary location. 

57 ''' 

58 raise NotImplementedError 

59 

60 

61# Private registry mapping URL schemes to source handlers 

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

63 __.accret.Dictionary( ) ) 

64 

65 

66def register_source_handler( 

67 handler: __.typx.Annotated[ 

68 AbstractSourceHandler, 

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

70 ], 

71 schemes: __.typx.Annotated[ 

72 __.cabc.Iterable[ str ], 

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

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

75 ] 

76) -> None: 

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

78 for scheme in schemes: 

79 _SCHEME_HANDLERS[ scheme ] = handler 

80 

81 

82def source_handler( 

83 schemes: __.typx.Annotated[ 

84 __.cabc.Iterable[ str ], 

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

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

87 ] 

88) -> __.cabc.Callable[ 

89 [ type[ AbstractSourceHandler ] ], type[ AbstractSourceHandler ] 

90]: 

91 ''' Decorator for automatic source handler registration. 

92 

93 Usage: 

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

95 class GitSourceHandler: 

96 ... 

97 ''' 

98 def decorator( 

99 handler_class: type[ AbstractSourceHandler ] 

100 ) -> type[ AbstractSourceHandler ]: 

101 register_source_handler( handler_class( ), schemes ) 

102 return handler_class 

103 return decorator 

104 

105 

106def resolve_source_location( 

107 source_spec: str, 

108 tag_prefix: __.typx.Annotated[ 

109 __.Absential[ str ], 

110 __.ddoc.Doc( 

111 "Prefix for filtering version tags when no explicit ref " 

112 "is specified. Only tags starting with this prefix will be " 

113 "considered, and the prefix will be stripped before version " 

114 "parsing." ), 

115 ] = __.absent, 

116) -> __.Path: 

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

118 

119 Delegates to registered source handlers based on URL scheme. 

120 Raises DataSourceNoSupport if no handler can process the specification. 

121 ''' 

122 for scheme, handler in _SCHEME_HANDLERS.items( ): 

123 if source_spec.startswith( scheme ): 

124 return handler.resolve( source_spec, tag_prefix ) 

125 raise __.DataSourceNoSupport( source_spec )