Coverage for src/lexigram/admin/services/storage/service.py: 45%

1""" 

2Main storage service implementation. 

3 

4Provides the AdminStorageService class for file upload management. 

5""" 

6 

7from __future__ import annotations 

8 

9from collections.abc import AsyncIterator 

10from datetime import UTC, datetime 

11import hashlib 

12from typing import Any, BinaryIO 

13 

14from lexigram.admin.config import AdminStorageConfig 

15from lexigram.admin.exceptions import DataError 

16from lexigram.admin.services.storage.types import ( 

17 AdminFileInfo, 

18 AdminUploadOptions, 

19) 

20from lexigram.admin.services.storage.utils import get_content_type 

21from lexigram.contracts.infra.storage import ( 

22 BlobStoreProtocol, 

23 FileInfo, 

24 StorageUnsupportedOperationError, 

25) 

26from lexigram.di.decorators import inject 

27from lexigram.logging import get_logger 

28from lexigram.result import Err, Ok, Result 

29 

30logger = get_logger(__name__) 

31 

32 

33@inject 

34class AdminStorageService: 

35 """Storage service for admin file management. 

36 

37 Provides a unified interface for file uploads with: 

38 - Validation (size, type) 

39 - Path generation 

40 - Presigned URLs 

41 - Resource association 

42 

43 Example: 

44 >>> storage = AdminStorageService(blob_store, config) 

45 >>> 

46 >>> # Upload a file 

47 >>> result = await storage.upload( 

48 ... data=file_bytes, 

49 ... filename="avatar.jpg", 

50 ... options=AdminUploadOptions( 

51 ... resource_type="users", 

52 ... resource_id=123, 

53 ... ), 

54 ... ) 

55 >>> 

56 >>> # Get presigned URL 

57 >>> url = await storage.get_download_url(result.file_info.path) 

58 """ 

59 

60 def __init__( 

61 self, 

62 blob_store: BlobStoreProtocol, 

63 config: AdminStorageConfig | None = None, 

64 ): 

65 self.config = config or AdminStorageConfig() 

66 self._store: BlobStoreProtocol = blob_store 

67 

68 def validate_upload( 

69 self, 

70 data: bytes | BinaryIO, 

71 filename: str, 

72 options: AdminUploadOptions | None = None, 

73 ) -> tuple[bool, str | None]: 

74 """Validate file before upload. 

75 

76 Args: 

77 data: File content 

78 filename: Original filename 

79 options: Upload options 

80 

81 Returns: 

82 (is_valid, error_message) 

83 """ 

84 # Get content type 

85 content_type = options.content_type if options else None 

86 if not content_type: 

87 content_type = get_content_type(filename) 

88 

89 # Check content type 

90 allowed_types = ( 

91 options.allowed_types 

92 if options and options.allowed_types 

93 else self.config.allowed_content_types 

94 ) 

95 if allowed_types and content_type not in allowed_types: 

96 return False, f"Content type not allowed: {content_type}" 

97 

98 # Check size 

99 max_size = ( 

100 options.max_size 

101 if options and options.max_size 

102 else self.config.max_file_size 

103 ) 

104 if isinstance(data, bytes): 

105 size = len(data) 

106 elif hasattr(data, "seek") and hasattr(data, "tell"): 

107 pos = data.tell() 

108 data.seek(0, 2) 

109 size = data.tell() 

110 data.seek(pos) 

111 else: 

112 size = 0 # Can't determine 

113 

114 if max_size and size > max_size: 

115 return False, f"File too large: {size} bytes (max: {max_size})" 

116 

117 return True, None 

118 

119 async def upload( 

120 self, 

121 data: bytes | BinaryIO | AsyncIterator[bytes], 

122 filename: str, 

123 options: AdminUploadOptions | None = None, 

124 uploaded_by: Any = None, 

125 ) -> Result[AdminFileInfo, DataError]: 

126 """Upload a file. 

127 

128 Args: 

129 data: File content 

130 filename: Original filename 

131 options: Upload options 

132 uploaded_by: User who uploaded the file 

133 

134 Returns: 

135 Result containing AdminFileInfo on success, DataError on failure. 

136 """ 

137 # Validate (only for bytes/BinaryIO) 

138 if isinstance(data, (bytes, BinaryIO)) or hasattr(data, "read"): 

139 is_valid, error = self.validate_upload(data, filename, options) # type: ignore[arg-type] 

140 if not is_valid: 

141 return Err(DataError(message=error or "Validation failed")) 

142 

143 # Generate path 

144 resource_type = options.resource_type if options else None 

145 resource_id = options.resource_id if options else None 

146 

147 from lexigram.admin.services.storage.utils import generate_upload_path 

148 

149 path = generate_upload_path( 

150 filename, 

151 resource_type=resource_type, 

152 resource_id=resource_id, 

153 base_path=self.config.base_path, 

154 ) 

155 

156 try: 

157 # Compute SHA-256 hash for bytes content 

158 sha256_hash: str | None = None 

159 if isinstance(data, bytes): 

160 sha256_hash = hashlib.sha256(data).hexdigest() 

161 

162 # Upload to storage 

163 upload_options = options.to_storage_options() if options else None 

164 if ( 

165 sha256_hash 

166 and upload_options is not None 

167 and hasattr(upload_options, "metadata") 

168 ): 

169 upload_options.metadata = { # type: ignore[misc] 

170 **(upload_options.metadata or {}), 

171 "sha256": sha256_hash, 

172 } 

173 result = await self._store.upload(path, data, upload_options) # type: ignore[arg-type] 

174 

175 # Convert to AdminFileInfo if needed 

176 file_info: AdminFileInfo 

177 if isinstance(result, AdminFileInfo): 

178 file_info = result 

179 elif FileInfo and isinstance(result, FileInfo): # type: ignore[truthy-function] 

180 file_info = AdminFileInfo( 

181 path=result.path, 

182 size=result.size, 

183 content_type=result.content_type, 

184 last_modified=result.last_modified, 

185 etag=result.etag, 

186 metadata=result.metadata, 

187 ) 

188 else: 

189 file_info = AdminFileInfo( 

190 path=path, 

191 size=len(data) if isinstance(data, bytes) else 0, 

192 content_type=get_content_type(filename), 

193 last_modified=datetime.now(UTC), 

194 ) 

195 

196 # Add admin metadata 

197 file_info.uploaded_by = uploaded_by 

198 file_info.resource_type = resource_type 

199 file_info.resource_id = resource_id 

200 

201 return Ok(file_info) 

202 except (RuntimeError, OSError, ConnectionError) as e: 

203 return Err(DataError(message=str(e), original_error=e)) 

204 

205 async def download(self, path: str) -> bytes: 

206 """Download file content. 

207 

208 Args: 

209 path: Storage path 

210 

211 Returns: 

212 File content as bytes 

213 """ 

214 return await self._store.download(path) 

215 

216 async def delete(self, path: str) -> bool: 

217 """Delete a file. 

218 

219 Args: 

220 path: Storage path 

221 

222 Returns: 

223 True if deleted 

224 """ 

225 try: 

226 await self._store.delete(path) 

227 return True 

228 except (RuntimeError, OSError): 

229 return False 

230 

231 async def exists(self, path: str) -> bool: 

232 """Check if file exists. 

233 

234 Args: 

235 path: Storage path 

236 

237 Returns: 

238 True if exists 

239 """ 

240 return await self._store.exists(path) 

241 

242 async def get_download_url( 

243 self, 

244 path: str, 

245 expires_in: int | None = None, 

246 ) -> str: 

247 """Get presigned download URL. 

248 

249 Args: 

250 path: Storage path 

251 expires_in: URL expiry in seconds 

252 

253 Returns: 

254 Presigned URL 

255 """ 

256 from datetime import timedelta 

257 

258 expires_seconds = expires_in or self.config.presigned_url_expiry 

259 try: 

260 return await self._store.get_presigned_url( 

261 path, 

262 expires_in=timedelta(seconds=expires_seconds), 

263 method="GET", 

264 ) 

265 except StorageUnsupportedOperationError: 

266 logger.warning( 

267 "storage.presigned_fallback_to_public", 

268 path=path, 

269 method="GET", 

270 ) 

271 return await self._store.get_url(path) 

272 

273 async def get_upload_url( 

274 self, 

275 filename: str, 

276 content_type: str | None = None, 

277 expires_in: int | None = None, 

278 resource_type: str | None = None, 

279 resource_id: Any = None, 

280 ) -> tuple[str, str]: 

281 """Get presigned upload URL for direct uploads. 

282 

283 Allows clients to upload directly to storage without 

284 going through the server. 

285 

286 Args: 

287 filename: Target filename 

288 

289 Returns: 

290 Tuple of (presigned_url, storage_path) 

291 """ 

292 from datetime import timedelta 

293 

294 expires_seconds = expires_in or self.config.presigned_url_expiry 

295 

296 from lexigram.admin.services.storage.utils import generate_upload_path 

297 

298 path = generate_upload_path( 

299 filename, 

300 resource_type=resource_type, 

301 resource_id=resource_id, 

302 base_path=self.config.base_path, 

303 ) 

304 

305 try: 

306 url = await self._store.get_presigned_url( 

307 path, 

308 expires_in=timedelta(seconds=expires_seconds), 

309 method="PUT", 

310 ) 

311 except StorageUnsupportedOperationError: 

312 logger.warning( 

313 "storage.presigned_fallback_to_public", 

314 path=path, 

315 method="PUT", 

316 ) 

317 url = await self._store.get_url(path) 

318 

319 return url, path