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

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 

60def register_source_handler( 

61 handler: __.typx.Annotated[ 

62 AbstractSourceHandler, 

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

64 ], 

65 schemes: __.typx.Annotated[ 

66 __.cabc.Iterable[ str ], 

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

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

69 ] 

70) -> None: 

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

72 for scheme in schemes: 

73 _SCHEME_HANDLERS[ scheme ] = handler 

74 

75 

76def source_handler( 

77 schemes: __.typx.Annotated[ 

78 __.cabc.Iterable[ str ], 

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

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

81 ] 

82) -> __.cabc.Callable[ 

83 [ type[ AbstractSourceHandler ] ], type[ AbstractSourceHandler ] 

84]: 

85 ''' Decorator for automatic source handler registration. 

86 

87 Usage: 

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

89 class GitSourceHandler: 

90 ... 

91 ''' 

92 def decorator( 

93 handler_class: type[ AbstractSourceHandler ] 

94 ) -> type[ AbstractSourceHandler ]: 

95 register_source_handler( handler_class( ), schemes ) 

96 return handler_class 

97 return decorator 

98 

99 

100def resolve_source_location( 

101 source_spec: str, 

102 tag_prefix: _nomina.TagPrefixArgument = __.absent, 

103) -> __.Path: 

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

105 

106 Delegates to registered source handlers based on URL scheme. 

107 Uses urlparse to extract the scheme from the specification. 

108 

109 Raises DataSourceNoSupport if no handler can process the specification. 

110 ''' 

111 if source_spec.startswith( 'git@' ): 

112 if 'git@' in _SCHEME_HANDLERS: 

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

114 source_spec, tag_prefix ) 

115 raise __.DataSourceNoSupport( source_spec ) 

116 parsed = __.urlparse.urlparse( source_spec ) 

117 if parsed.scheme in _SCHEME_HANDLERS: 

118 return _SCHEME_HANDLERS[ parsed.scheme ].resolve( 

119 source_spec, tag_prefix ) 

120 raise __.DataSourceNoSupport( source_spec )