Coverage for src / lexigram / admin / services / storage / upload.py: 31%

1"""File Upload Service - Refactored to use lexigram-storage. 

2 

3This service replaces the custom LocalStorageBackend with lexigram-storage 

4for S3/Azure/GCS 

5File upload service with validation and storage. 

6 

7Provides a clean API for handling file uploads with validation, 

8storage backends, and security checks. 

9""" 

10 

11from __future__ import annotations 

12 

13from dataclasses import dataclass 

14from enum import StrEnum 

15import hashlib 

16from pathlib import Path 

17from typing import BinaryIO 

18import uuid 

19 

20from lexigram.contracts.infra.storage import BlobStoreProtocol, FileInfo, UploadOptions 

21from lexigram.di.decorators import inject 

22 

23 

24class StorageBackend(StrEnum): 

25 """Supported storage backends.""" 

26 

27 LOCAL = "local" 

28 S3 = "s3" 

29 AZURE = "azure" 

30 GCS = "gcs" 

31 MEMORY = "memory" # For testing 

32 

33 

34@dataclass 

35class UploadedFile: 

36 """Uploaded file metadata.""" 

37 

38 filename: str 

39 content_type: str 

40 size: int 

41 storage_path: str 

42 url: str 

43 hash: str = "" 

44 

45 

46@inject 

47class FileValidator: 

48 """File upload validator.""" 

49 

50 def __init__( 

51 self, 

52 max_size: int = 10 * 1024 * 1024, # 10MB 

53 allowed_extensions: set[str] | None = None, 

54 allowed_mimetypes: set[str] | None = None, 

55 ): 

56 """ 

57 Initialize validator. 

58 

59 Args: 

60 max_size: Maximum file size in bytes 

61 allowed_extensions: Set of allowed file extensions (e.g., {'.jpg', '.png'}) 

62 allowed_mimetypes: Set of allowed MIME types 

63 """ 

64 self.max_size = max_size 

65 self.allowed_extensions = allowed_extensions or set() 

66 self.allowed_mimetypes = allowed_mimetypes or set() 

67 

68 def validate(self, filename: str, size: int, content_type: str) -> tuple[bool, str]: 

69 """ 

70 Validate file. 

71 

72 Returns: 

73 (is_valid, error_message) 

74 """ 

75 # Check size 

76 if size > self.max_size: 

77 return False, f"File too large ({size} bytes, max {self.max_size})" 

78 

79 # Check extension 

80 if self.allowed_extensions: 

81 ext = Path(filename).suffix.lower() 

82 if ext not in self.allowed_extensions: 

83 return False, f"File type not allowed: {ext}" 

84 

85 # Check MIME type 

86 if self.allowed_mimetypes and content_type not in self.allowed_mimetypes: 

87 return False, f"MIME type not allowed: {content_type}" 

88 

89 return True, "" 

90 

91 

92@inject 

93class FileUploadService: 

94 """DI-injectable file upload service using lexigram-storage. 

95 

96 Benefits over custom LocalStorageBackend: 

97 - S3/Azure/GCS support out of the box 

98 - Presigned URLs for secure downloads 

99 - Production-tested storage patterns 

100 - Automatic retry logic 

101 - Cloud-native optimizations 

102 """ 

103 

104 def __init__( 

105 self, 

106 storage: BlobStoreProtocol, 

107 validator: FileValidator | None = None, 

108 upload_prefix: str = "admin", 

109 ): 

110 """ 

111 Initialize file upload service. 

112 

113 Args: 

114 storage: BlobStoreProtocol instance (injected) 

115 validator: File validator (defaults to permissive validator) 

116 upload_prefix: Prefix for uploaded files (e.g., "admin/uploads") 

117 """ 

118 self.storage = storage 

119 self.validator = validator or FileValidator() 

120 self.upload_prefix = upload_prefix 

121 

122 async def upload( 

123 self, 

124 file: BinaryIO, 

125 filename: str, 

126 content_type: str = "application/octet-stream", 

127 public: bool = False, 

128 ) -> tuple[UploadedFile | None, str]: 

129 """ 

130 Upload file using lexigram-storage. 

131 

132 Args: 

133 file: File-like object 

134 filename: Original filename 

135 content_type: MIME type 

136 public: Whether file should be publicly accessible 

137 

138 Returns: 

139 (UploadedFile, error_message) - UploadedFile is None if validation failed 

140 """ 

141 # Read file content for validation and hashing 

142 content = file.read() 

143 size = len(content) 

144 

145 # Validate 

146 is_valid, error = self.validator.validate(filename, size, content_type) 

147 if not is_valid: 

148 return None, error 

149 

150 # Generate unique filename 

151 ext = Path(filename).suffix 

152 unique_name = f"{uuid.uuid4().hex}{ext}" 

153 storage_path = f"{self.upload_prefix}/{unique_name}" 

154 

155 # Calculate hash 

156 hasher = hashlib.sha256() 

157 hasher.update(content) 

158 file_hash = hasher.hexdigest() 

159 

160 try: 

161 # Upload using lexigram-storage 

162 _file_info: FileInfo = await self.storage.upload( 

163 data=content, 

164 path=storage_path, 

165 options=UploadOptions( 

166 content_type=content_type, 

167 public=public, 

168 metadata={"original_filename": filename, "sha256": file_hash}, 

169 ), 

170 ) 

171 

172 # Get URL (presigned if private, public if public) 

173 if public: 

174 url = await self.storage.get_url(storage_path) 

175 else: 

176 from datetime import timedelta 

177 

178 url = await self.storage.get_presigned_url( 

179 path=storage_path, 

180 expires_in=timedelta(hours=1), 

181 ) 

182 

183 return ( 

184 UploadedFile( 

185 filename=filename, 

186 content_type=content_type, 

187 size=size, 

188 storage_path=storage_path, 

189 url=url, 

190 hash=file_hash, 

191 ), 

192 "", 

193 ) 

194 

195 except (OSError, RuntimeError, AttributeError): 

196 from lexigram.logging import get_logger 

197 

198 logger = get_logger(__name__) 

199 # Emit an explicit ERROR-level text message for caplog tests 

200 logger.exception("Upload failed for %s", filename) 

201 # Keep stacktrace for diagnostics 

202 logger.exception("Upload failed for %s", filename) 

203 return None, "Upload failed" 

204 except BaseException: 

205 from lexigram.logging import get_logger 

206 

207 logger = get_logger(__name__) 

208 logger.exception("Unexpected upload failure for %s", filename) 

209 logger.exception("Unexpected upload failure for %s", filename) 

210 return None, "Upload failed" 

211 

212 async def delete(self, storage_path: str) -> bool: 

213 """ 

214 Delete uploaded file. 

215 

216 Args: 

217 storage_path: Path to file in storage 

218 

219 Returns: 

220 True if deleted successfully 

221 """ 

222 try: 

223 await self.storage.delete(storage_path) 

224 return True 

225 except (OSError, RuntimeError): 

226 from lexigram.logging import get_logger 

227 

228 logger = get_logger(__name__) 

229 logger.exception("Failed to delete storage path %s", storage_path) 

230 logger.exception("Failed to delete storage path %s", storage_path) 

231 return False 

232 except BaseException: 

233 from lexigram.logging import get_logger 

234 

235 logger = get_logger(__name__) 

236 logger.exception("Unexpected error deleting storage path %s", storage_path) 

237 logger.exception("Unexpected error deleting storage path %s", storage_path) 

238 return False 

239 

240 async def get_url(self, storage_path: str, expires_in: int = 3600) -> str: 

241 """ 

242 Get presigned URL for file. 

243 

244 Args: 

245 storage_path: Path to file in storage 

246 expires_in: URL expiration time in seconds 

247 

248 Returns: 

249 Presigned URL 

250 """ 

251 from datetime import timedelta 

252 

253 return await self.storage.get_presigned_url( 

254 path=storage_path, 

255 expires_in=timedelta(seconds=expires_in), 

256 ) 

257 

258 async def exists(self, storage_path: str) -> bool: 

259 """ 

260 Check if file exists. 

261 

262 Args: 

263 storage_path: Path to file in storage 

264 

265 Returns: 

266 True if file exists 

267 """ 

268 try: 

269 await self.storage.info(storage_path) 

270 return True 

271 except (OSError, RuntimeError): 

272 from lexigram.logging import get_logger 

273 

274 logger = get_logger(__name__) 

275 logger.exception("Failed to check existence for %s", storage_path) 

276 logger.exception("Failed to check existence for %s", storage_path) 

277 return False 

278 except BaseException: 

279 from lexigram.logging import get_logger 

280 

281 logger = get_logger(__name__) 

282 logger.exception("Unexpected error checking existence for %s", storage_path) 

283 logger.exception("Unexpected error checking existence for %s", storage_path) 

284 return False