Coverage for src/ezcompiler/adapters/uploader_factory.py: 60.00%

1# /////////////////////////////////////////////////////////////// 

2# UPLOADER_FACTORY - Uploader factory for creating uploader instances 

3# Project: ezcompiler 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7Uploader factory - Factory for creating uploader instances. 

8 

9This module provides a centralized factory for creating and configuring 

10uploader instances based on type and configuration, with support for 

11validation and discovery of supported types. 

12""" 

13 

14from __future__ import annotations 

15 

16# /////////////////////////////////////////////////////////////// 

17# IMPORTS 

18# /////////////////////////////////////////////////////////////// 

19# Standard library imports 

20from typing import Any 

21 

22from .._types import UploaderPort 

23 

24# Local imports 

25from ..shared.exceptions import UploadError 

26from ..utils import UploaderUtils 

27from ._disk_uploader import DiskUploader 

28from ._r2_uploader import R2Uploader 

29from ._server_uploader import ServerUploader 

30 

31# /////////////////////////////////////////////////////////////// 

32# CLASSES 

33# /////////////////////////////////////////////////////////////// 

34 

35 

36class UploaderFactory: 

37 """ 

38 Factory class for creating uploader instances. 

39 

40 Provides a centralized way to create and configure uploader instances 

41 based on type specification, with support for validation and type discovery. 

42 

43 Example: 

44 >>> uploader = UploaderFactory.create_uploader("disk", {"overwrite": True}) 

45 >>> types = UploaderFactory.get_supported_types() 

46 >>> print(types) 

47 ['disk', 'server'] 

48 """ 

49 

50 # //////////////////////////////////////////////// 

51 # FACTORY METHODS 

52 # //////////////////////////////////////////////// 

53 

54 @staticmethod 

55 def create_uploader( 

56 upload_type: str, config: dict[str, Any] | None = None 

57 ) -> UploaderPort: 

58 """ 

59 Create an uploader instance based on the specified type. 

60 

61 Args: 

62 upload_type: Type of uploader ("disk" or "server") 

63 config: Configuration dictionary for the uploader (default: None) 

64 

65 Returns: 

66 UploaderPort: Configured uploader instance (satisfies the Port) 

67 

68 Raises: 

69 UploadError: If upload type is not supported 

70 

71 Example: 

72 >>> disk_uploader = UploaderFactory.create_uploader("disk") 

73 >>> server_uploader = UploaderFactory.create_uploader( 

74 ... "server", {"server_url": "https://example.com"} 

75 ... ) 

76 """ 

77 upload_type = upload_type.lower() 

78 

79 # Validate upload type using UploaderUtils 

80 UploaderUtils.validate_upload_type(upload_type) 

81 

82 if upload_type == "disk": 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 return DiskUploader(config) 

84 elif upload_type == "server": 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true

85 return ServerUploader(config) 

86 elif upload_type == "r2": 86 ↛ 90line 86 didn't jump to line 90 because the condition on line 86 was always true

87 return R2Uploader(config) 

88 else: 

89 # This should not happen due to validation above, but kept for safety 

90 raise UploadError(f"Unsupported upload type: {upload_type}") 

91 

92 @staticmethod 

93 def create_from_config(config: dict[str, Any]) -> UploaderPort: 

94 """ 

95 Create an uploader instance from a configuration dictionary. 

96 

97 Args: 

98 config: Configuration dictionary containing: 

99 - type: Upload type ("disk" or "server") 

100 - config: Uploader-specific configuration (optional) 

101 

102 Returns: 

103 UploaderPort: Configured uploader instance (satisfies the Port) 

104 

105 Raises: 

106 UploadError: If configuration is invalid or type is missing 

107 

108 Example: 

109 >>> config = { 

110 ... "type": "disk", 

111 ... "config": {"overwrite": True} 

112 ... } 

113 >>> uploader = UploaderFactory.create_from_config(config) 

114 """ 

115 if "type" not in config: 

116 raise UploadError("Upload type not specified in configuration") 

117 

118 upload_type = config["type"] 

119 uploader_config = config.get("config", {}) 

120 

121 return UploaderFactory.create_uploader(upload_type, uploader_config) 

122 

123 # //////////////////////////////////////////////// 

124 # UTILITY METHODS 

125 # //////////////////////////////////////////////// 

126 

127 @staticmethod 

128 def get_supported_types() -> list[str]: 

129 """ 

130 Get list of supported upload types. 

131 

132 Returns: 

133 list[str]: List of supported upload type names 

134 

135 Example: 

136 >>> types = UploaderFactory.get_supported_types() 

137 >>> print(types) 

138 ['disk', 'server', 'r2'] 

139 """ 

140 return ["disk", "server", "r2"] 

141 

142 @staticmethod 

143 def validate_config(upload_type: str, config: dict[str, Any] | None = None) -> bool: 

144 """ 

145 Validate configuration for a specific upload type. 

146 

147 Args: 

148 upload_type: Type of uploader to validate 

149 config: Configuration to validate (default: None) 

150 

151 Returns: 

152 bool: True if configuration is valid, False otherwise 

153 

154 Note: 

155 Creates a temporary uploader instance to test configuration validity. 

156 

157 Example: 

158 >>> is_valid = UploaderFactory.validate_config("disk", {"overwrite": True}) 

159 >>> print(is_valid) 

160 True 

161 """ 

162 try: 

163 UploaderFactory.create_uploader(upload_type, config) 

164 return True 

165 except Exception: 

166 return False