Coverage for src / lexigram / admin / services / storage / service.py: 26%
85 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""
2Main storage service implementation.
4Provides the AdminStorageService class for file upload management.
5"""
7from __future__ import annotations
9from collections.abc import AsyncIterator
10from datetime import UTC, datetime
11import hashlib
12from typing import Any, BinaryIO
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 BlobStoreProtocol, FileInfo
22from lexigram.di.decorators import inject
23from lexigram.result import Err, Ok, Result
26@inject
27class AdminStorageService:
28 """Storage service for admin file management.
30 Provides a unified interface for file uploads with:
31 - Validation (size, type)
32 - Path generation
33 - Presigned URLs
34 - Resource association
36 Example:
37 >>> storage = AdminStorageService(blob_store, config)
38 >>>
39 >>> # Upload a file
40 >>> result = await storage.upload(
41 ... data=file_bytes,
42 ... filename="avatar.jpg",
43 ... options=AdminUploadOptions(
44 ... resource_type="users",
45 ... resource_id=123,
46 ... ),
47 ... )
48 >>>
49 >>> # Get presigned URL
50 >>> url = await storage.get_download_url(result.file_info.path)
51 """
53 def __init__(
54 self,
55 blob_store: BlobStoreProtocol,
56 config: AdminStorageConfig | None = None,
57 ):
58 self.config = config or AdminStorageConfig()
59 self._store: BlobStoreProtocol = blob_store
61 def validate_upload(
62 self,
63 data: bytes | BinaryIO,
64 filename: str,
65 options: AdminUploadOptions | None = None,
66 ) -> tuple[bool, str | None]:
67 """Validate file before upload.
69 Args:
70 data: File content
71 filename: Original filename
72 options: Upload options
74 Returns:
75 (is_valid, error_message)
76 """
77 # Get content type
78 content_type = options.content_type if options else None
79 if not content_type:
80 content_type = get_content_type(filename)
82 # Check content type
83 allowed_types = (
84 options.allowed_types
85 if options and options.allowed_types
86 else self.config.allowed_content_types
87 )
88 if allowed_types and content_type not in allowed_types:
89 return False, f"Content type not allowed: {content_type}"
91 # Check size
92 max_size = (
93 options.max_size
94 if options and options.max_size
95 else self.config.max_file_size
96 )
97 if isinstance(data, bytes):
98 size = len(data)
99 elif hasattr(data, "seek") and hasattr(data, "tell"):
100 pos = data.tell()
101 data.seek(0, 2)
102 size = data.tell()
103 data.seek(pos)
104 else:
105 size = 0 # Can't determine
107 if max_size and size > max_size:
108 return False, f"File too large: {size} bytes (max: {max_size})"
110 return True, None
112 async def upload(
113 self,
114 data: bytes | BinaryIO | AsyncIterator[bytes],
115 filename: str,
116 options: AdminUploadOptions | None = None,
117 uploaded_by: Any = None,
118 ) -> Result[AdminFileInfo, DataError]:
119 """Upload a file.
121 Args:
122 data: File content
123 filename: Original filename
124 options: Upload options
125 uploaded_by: User who uploaded the file
127 Returns:
128 Result containing AdminFileInfo on success, DataError on failure.
129 """
130 # Validate (only for bytes/BinaryIO)
131 if isinstance(data, (bytes, BinaryIO)) or hasattr(data, "read"):
132 is_valid, error = self.validate_upload(data, filename, options) # type: ignore[arg-type]
133 if not is_valid:
134 return Err(DataError(message=error or "Validation failed"))
136 # Generate path
137 resource_type = options.resource_type if options else None
138 resource_id = options.resource_id if options else None
140 from lexigram.admin.services.storage.utils import generate_upload_path
142 path = generate_upload_path(
143 filename,
144 resource_type=resource_type,
145 resource_id=resource_id,
146 base_path=self.config.base_path,
147 )
149 try:
150 # Compute SHA-256 hash for bytes content
151 sha256_hash: str | None = None
152 if isinstance(data, bytes):
153 sha256_hash = hashlib.sha256(data).hexdigest()
155 # Upload to storage
156 upload_options = options.to_storage_options() if options else None
157 if (
158 sha256_hash
159 and upload_options is not None
160 and hasattr(upload_options, "metadata")
161 ):
162 upload_options.metadata = { # type: ignore[misc]
163 **(upload_options.metadata or {}),
164 "sha256": sha256_hash,
165 }
166 result = await self._store.upload(path, data, upload_options) # type: ignore[arg-type]
168 # Convert to AdminFileInfo if needed
169 file_info: AdminFileInfo
170 if isinstance(result, AdminFileInfo):
171 file_info = result
172 elif FileInfo and isinstance(result, FileInfo): # type: ignore[truthy-function]
173 file_info = AdminFileInfo(
174 path=result.path,
175 size=result.size,
176 content_type=result.content_type,
177 last_modified=result.last_modified,
178 etag=result.etag,
179 metadata=result.metadata,
180 )
181 else:
182 file_info = AdminFileInfo(
183 path=path,
184 size=len(data) if isinstance(data, bytes) else 0,
185 content_type=get_content_type(filename),
186 last_modified=datetime.now(UTC),
187 )
189 # Add admin metadata
190 file_info.uploaded_by = uploaded_by
191 file_info.resource_type = resource_type
192 file_info.resource_id = resource_id
194 return Ok(file_info)
195 except (RuntimeError, OSError, ConnectionError) as e:
196 return Err(DataError(message=str(e), original_error=e))
198 async def download(self, path: str) -> bytes:
199 """Download file content.
201 Args:
202 path: Storage path
204 Returns:
205 File content as bytes
206 """
207 return await self._store.download(path)
209 async def delete(self, path: str) -> bool:
210 """Delete a file.
212 Args:
213 path: Storage path
215 Returns:
216 True if deleted
217 """
218 try:
219 await self._store.delete(path)
220 return True
221 except (RuntimeError, OSError):
222 return False
224 async def exists(self, path: str) -> bool:
225 """Check if file exists.
227 Args:
228 path: Storage path
230 Returns:
231 True if exists
232 """
233 return await self._store.exists(path)
235 async def get_download_url(
236 self,
237 path: str,
238 expires_in: int | None = None,
239 ) -> str:
240 """Get presigned download URL.
242 Args:
243 path: Storage path
244 expires_in: URL expiry in seconds
246 Returns:
247 Presigned URL
248 """
249 from datetime import timedelta
251 expires_seconds = expires_in or self.config.presigned_url_expiry
252 return await self._store.get_presigned_url(
253 path,
254 expires_in=timedelta(seconds=expires_seconds),
255 method="GET",
256 )
258 async def get_upload_url(
259 self,
260 filename: str,
261 content_type: str | None = None,
262 expires_in: int | None = None,
263 resource_type: str | None = None,
264 resource_id: Any = None,
265 ) -> tuple[str, str]:
266 """Get presigned upload URL for direct uploads.
268 Allows clients to upload directly to storage without
269 going through the server.
271 Args:
272 filename: Target filename
274 Returns:
275 Tuple of (presigned_url, storage_path)
276 """
277 from datetime import timedelta
279 expires_seconds = expires_in or self.config.presigned_url_expiry
281 from lexigram.admin.services.storage.utils import generate_upload_path
283 path = generate_upload_path(
284 filename,
285 resource_type=resource_type,
286 resource_id=resource_id,
287 base_path=self.config.base_path,
288 )
290 url = await self._store.get_presigned_url(
291 path,
292 expires_in=timedelta(seconds=expires_seconds),
293 method="PUT",
294 )
296 return url, path