Coverage for sources/librovore/xtnsmgr/installation.py: 60%

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''' Async package installation for extensions. ''' 

22 

23 

24import uv as _uv 

25 

26from . import __ 

27 

28 

29_scribe = __.acquire_scribe( __name__ ) 

30 

31 

32async def install_package( 

33 specification: str, 

34 cache_path: __.Path, *, 

35 retries_max: int = 3 

36) -> __.Path: 

37 ''' Installs package to specified path with retry logic. 

38 

39 Returns path to installed package for sys.path manipulation. 

40 ''' 

41 for attempt in range( retries_max + 1 ): 41 ↛ 54line 41 didn't jump to line 54 because the loop on line 41 didn't complete

42 try: return await _install_with_uv( specification, cache_path ) 

43 except __.ExtensionInstallFailure as exc: # noqa: PERF203 

44 if attempt == retries_max: 

45 _scribe.error( 

46 f"Failed to install {specification} after " 

47 f"{retries_max + 1} attempts: {exc}" ) 

48 raise 

49 delay = 2 ** attempt 

50 _scribe.warning( 

51 f"Installation attempt {attempt + 1} failed for " 

52 f"{specification}, retrying in {delay}s: {exc}" ) 

53 await __.asyncio.sleep( delay ) 

54 raise __.ExtensionInstallFailure( 

55 specification, "Maximum retries exceeded" ) 

56 

57 

58async def _install_with_uv( 

59 specification: str, cache_path: __.Path 

60) -> __.Path: 

61 ''' Installs package using uv to specified directory. ''' 

62 cache_path.mkdir( parents = True, exist_ok = True ) 

63 executable = _get_uv_executable( specification ) 

64 command = _build_uv_command( executable, cache_path, specification ) 

65 _scribe.info( f"Installing {specification} to {cache_path}." ) 

66 stdout, stderr, returncode = await _execute_uv_command( 

67 command, specification ) 

68 _validate_installation_result( specification, returncode, stderr ) 

69 _scribe.info( f"Successfully installed {specification}." ) 

70 return cache_path 

71 

72 

73 

74 

75def _get_uv_executable( specification: str ) -> str: 

76 ''' Gets uv executable path, raising appropriate error if not found. ''' 

77 try: return str( _uv.find_uv_bin( ) ) 

78 except ImportError as exc: 

79 raise __.ExtensionInstallFailure( 

80 specification, f"uv not available: {exc}" ) from exc 

81 

82 

83def _build_uv_command( 

84 executable: str, cache_path: __.Path, specification: str 

85) -> list[ str ]: 

86 ''' Builds uv command for package installation. ''' 

87 return [ 

88 executable, 'pip', 

89 'install', '--target', str( cache_path ), 

90 specification 

91 ] 

92 

93 

94async def _execute_uv_command( 

95 command: list[ str ], specification: str 

96) -> tuple[ bytes, bytes, int ]: 

97 ''' Executes uv command and returns stdout, stderr, and return code. ''' 

98 try: 

99 process = await __.asyncio.create_subprocess_exec( 

100 *command, 

101 stdout = __.asyncio.subprocess.PIPE, 

102 stderr = __.asyncio.subprocess.PIPE ) 

103 except OSError as exc: 

104 raise __.ExtensionInstallFailure( 

105 specification, f"Process execution failed: {exc}" ) from exc 

106 try: 

107 stdout, stderr = await process.communicate( ) 

108 except __.asyncio.TimeoutError as exc: 

109 raise __.ExtensionInstallFailure( 

110 specification, f"Installation timed out: {exc}" ) from exc 

111 returncode = process.returncode 

112 if returncode is None: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true

113 raise __.ExtensionInstallFailure( 

114 specification, "Process terminated without exit code" ) 

115 return stdout, stderr, returncode 

116 

117 

118def _validate_installation_result( 

119 specification: str, returncode: int, stderr: bytes 

120) -> None: 

121 ''' Validates installation result and raises error if failed. ''' 

122 if returncode != 0: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 raise __.ExtensionInstallFailure( 

124 specification, 

125 f"uv install failed (exit {returncode}): " 

126 f"{stderr.decode( 'utf-8' )}" ) 

127 

128 

129 

130