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( self, source_spec: str ) -> __.Path: 

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

42 

43 Returns path to directory containing the resolved source content. 

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

45 a temporary location. 

46 ''' 

47 raise NotImplementedError 

48 

49 

50# Private registry mapping URL schemes to source handlers 

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

52 __.accret.Dictionary( ) ) 

53 

54 

55def register_source_handler( 

56 handler: __.typx.Annotated[ 

57 AbstractSourceHandler, 

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

59 ], 

60 schemes: __.typx.Annotated[ 

61 __.cabc.Iterable[ str ], 

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

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

64 ] 

65) -> None: 

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

67 for scheme in schemes: 

68 _SCHEME_HANDLERS[ scheme ] = handler 

69 

70 

71def source_handler( 

72 schemes: __.typx.Annotated[ 

73 __.cabc.Iterable[ str ], 

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

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

76 ] 

77) -> __.cabc.Callable[ 

78 [ type[ AbstractSourceHandler ] ], type[ AbstractSourceHandler ] 

79]: 

80 ''' Decorator for automatic source handler registration. 

81 

82 Usage: 

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

84 class GitSourceHandler: 

85 ... 

86 ''' 

87 def decorator( 

88 handler_class: type[ AbstractSourceHandler ] 

89 ) -> type[ AbstractSourceHandler ]: 

90 register_source_handler( handler_class( ), schemes ) 

91 return handler_class 

92 return decorator 

93 

94 

95def resolve_source_location( source_spec: str ) -> __.Path: 

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

97 

98 Delegates to registered source handlers based on URL scheme. 

99 Raises DataSourceNoSupport if no handler can process the specification. 

100 ''' 

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

102 if source_spec.startswith( scheme ): 

103 return handler.resolve( source_spec ) 

104 raise __.DataSourceNoSupport( source_spec )