Coverage for src/lexigram/admin/actions/bulk_manager.py: 58%
152 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""
2Advanced Bulk Action Manager for Lexigram Admin.
4Provides comprehensive bulk operation support including:
5- Bulk editing and assignment
6- Custom action registration
7- Progress tracking
8- Undo functionality via snapshots
9"""
11from __future__ import annotations
13import asyncio
14from collections.abc import Callable
15from dataclasses import dataclass, field
16from datetime import datetime
17from typing import Any, Generic, Protocol, TypeVar
19from lexigram.admin.exceptions import AdminError, NotFoundError
20from lexigram.contracts.infra.cache import CacheBackendProtocol
21from lexigram.logging import get_logger
22from lexigram.result import Err, Ok, Result
24logger = get_logger(__name__)
26T = TypeVar("T")
29class IBulkDataSource(Protocol[T]):
30 """Protocol for data sources that support bulk operations."""
32 async def bulk_update(
33 self,
34 ids: list[Any],
35 updates: dict[str, Any],
36 ) -> int:
37 """Update multiple records."""
38 ...
40 async def fetch_by_ids(self, ids: list[Any]) -> list[T]:
41 """Fetch records by IDs."""
42 ...
44 async def create_snapshot(self, ids: list[Any]) -> str:
45 """Create a snapshot for undo."""
46 ...
48 async def restore_snapshot(self, snapshot_id: str) -> int:
49 """Restore from a snapshot."""
50 ...
53@dataclass
54class BulkEditField:
55 """Field configuration for bulk editing."""
57 name: str
58 label: str
59 field_type: str = "text"
60 options: list[tuple[Any, str]] | None = None
61 required: bool = False
62 validation: Callable[[Any], bool] | None = None
63 help_text: str | None = None
66@dataclass
67class BulkAssignConfig:
68 """Configuration for bulk assign operations."""
70 field_name: str
71 label: str
72 options: list[tuple[Any, str]]
73 allow_null: bool = False
74 confirm_message: str | None = None
77@dataclass
78class BulkActionResult:
79 """Payload of a completed bulk action operation.
81 Always carried inside ``Ok[BulkActionResult, AdminError]``. Per-item
82 failures are accumulated in ``errors``; a non-empty ``errors`` list does
83 NOT mean the overall operation failed — only an ``Err`` return does.
84 """
86 affected_count: int = 0
87 errors: list[str] = field(default_factory=list)
88 snapshot_id: str | None = None
89 duration_ms: float = 0.0
90 metadata: dict[str, Any] = field(default_factory=dict)
93@dataclass
94class BulkActionSnapshot:
95 """Snapshot of records before bulk action."""
97 snapshot_id: str
98 action_name: str
99 record_ids: list[Any]
100 timestamp: datetime
101 user_id: Any | None = None
102 metadata: dict[str, Any] = field(default_factory=dict)
105class BulkActionProgress:
106 """Track progress of long-running bulk actions."""
108 def __init__(self, total: int):
109 self.total = total
110 self.current = 0
111 self.errors: list[str] = []
112 self.start_time = datetime.now()
114 @property
115 def percentage(self) -> float:
116 """Get completion percentage."""
117 if self.total == 0:
118 return 100.0
119 return (self.current / self.total) * 100
121 @property
122 def elapsed_ms(self) -> float:
123 """Get elapsed time in milliseconds."""
124 delta = datetime.now() - self.start_time
125 return delta.total_seconds() * 1000
127 def increment(self, count: int = 1) -> Any:
128 """Increment progress counter."""
129 self.current += count
131 def add_error(self, error: str) -> Any:
132 """Add an error message."""
133 self.errors.append(error)
135 def to_dict(self) -> dict[str, Any]:
136 """Convert to dictionary for JSON serialization."""
137 return {
138 "total": self.total,
139 "current": self.current,
140 "percentage": self.percentage,
141 "elapsed_ms": self.elapsed_ms,
142 "errors": self.errors,
143 }
146class BulkActionManager(Generic[T]):
147 """Manager for advanced bulk operations on data tables."""
149 def __init__(
150 self,
151 data_source: IBulkDataSource[T],
152 cache: CacheBackendProtocol | None = None,
153 ):
154 self.data_source = data_source
155 self._cache_backend = cache
156 self._custom_actions: dict[str, Callable] = {}
157 self._snapshots: dict[str, BulkActionSnapshot] = {}
159 async def bulk_edit(
160 self,
161 ids: list[Any],
162 updates: dict[str, Any],
163 create_snapshot: bool = True,
164 batch_size: int = 100,
165 ) -> Result[BulkActionResult, AdminError]:
166 """Update multiple records with new field values."""
167 start_time = datetime.now()
168 snapshot_id = None
170 if create_snapshot:
171 snapshot_id = await self.data_source.create_snapshot(ids)
172 self._snapshots[snapshot_id] = BulkActionSnapshot(
173 snapshot_id=snapshot_id,
174 action_name="bulk_edit",
175 record_ids=ids,
176 timestamp=datetime.now(),
177 )
179 total_updated = 0
180 progress = BulkActionProgress(total=len(ids))
182 for i in range(0, len(ids), batch_size):
183 batch_ids = ids[i : i + batch_size]
184 count = await self.data_source.bulk_update(batch_ids, updates)
185 total_updated += count
186 progress.increment(len(batch_ids))
187 await asyncio.sleep(0)
189 duration = (datetime.now() - start_time).total_seconds() * 1000
191 return Ok(
192 BulkActionResult(
193 affected_count=total_updated,
194 snapshot_id=snapshot_id,
195 duration_ms=duration,
196 metadata={"updates": updates, "batch_size": batch_size},
197 )
198 )
200 async def bulk_assign(
201 self,
202 ids: list[Any],
203 field_name: str,
204 value: Any,
205 create_snapshot: bool = True,
206 ) -> Result[BulkActionResult, AdminError]:
207 """Assign a value to a specific field for multiple records."""
208 return await self.bulk_edit(
209 ids=ids,
210 updates={field_name: value},
211 create_snapshot=create_snapshot,
212 )
214 def register_action(
215 self,
216 name: str,
217 handler: Callable[[list[Any]], Result[BulkActionResult, AdminError]],
218 ) -> None:
219 """Register a custom bulk action."""
220 self._custom_actions[name] = handler
222 async def execute_action(
223 self,
224 action_name: str,
225 ids: list[Any],
226 ) -> Result[BulkActionResult, AdminError]:
227 """Execute a registered custom bulk action."""
228 if action_name not in self._custom_actions:
229 return Err(NotFoundError(f"Unknown action: {action_name}")) # type: ignore[arg-type]
231 handler = self._custom_actions[action_name]
232 return await handler(ids)
234 async def get_preview(
235 self,
236 ids: list[Any],
237 limit: int = 5,
238 ) -> list[T]:
239 """Get preview of records that will be affected."""
240 preview_ids = ids[:limit]
241 return await self.data_source.fetch_by_ids(preview_ids)
243 def get_confirmation_message(
244 self,
245 action_name: str,
246 count: int,
247 preview: list[T] | None = None,
248 ) -> str:
249 """Generate confirmation message for bulk action."""
250 msg = f"Are you sure you want to {action_name} {count} record(s)?"
252 if preview:
253 msg += "\n\nThis will affect:"
254 for record in preview:
255 msg += f"\n- {record}"
257 return msg
259 async def execute_with_progress(
260 self,
261 ids: list[Any],
262 handler: Callable[[Any, BulkActionProgress], Any],
263 batch_size: int = 10,
264 ) -> Result[BulkActionResult, AdminError]:
265 """Execute bulk action with progress tracking."""
266 start_time = datetime.now()
267 progress = BulkActionProgress(total=len(ids))
269 progress_key = f"bulk_progress_{id(progress)}"
270 if self._cache_backend is not None:
271 await self._cache_backend.set(progress_key, progress.to_dict(), 300)
273 for i, record_id in enumerate(ids):
274 try:
275 handler_result = handler(record_id, progress)
276 if handler_result is not None:
277 if hasattr(handler_result, "__await__"):
278 await handler_result
279 progress.increment()
280 except (
281 ValueError,
282 ConnectionError,
283 TimeoutError,
284 OSError,
285 KeyError,
286 ) as e:
287 progress.add_error(f"Error processing {record_id}: {e}")
289 if i % batch_size == 0:
290 if self._cache_backend is not None:
291 await self._cache_backend.set(progress_key, progress.to_dict(), 300)
292 await asyncio.sleep(0)
294 if self._cache_backend is not None:
295 await self._cache_backend.set(progress_key, progress.to_dict(), 300)
297 duration = (datetime.now() - start_time).total_seconds() * 1000
299 return Ok(
300 BulkActionResult(
301 affected_count=progress.current,
302 errors=progress.errors,
303 duration_ms=duration,
304 metadata={"progress_key": progress_key},
305 )
306 )
308 async def get_progress(self, progress_key: str) -> dict[str, Any] | None:
309 """Get current progress of a bulk action."""
310 if self._cache_backend is None:
311 return None
312 res = await self._cache_backend.get(progress_key)
313 return res.unwrap() if res.is_ok() else None
315 async def undo(self, snapshot_id: str) -> Result[BulkActionResult, AdminError]:
316 """Undo a previous bulk action by restoring from snapshot."""
317 if snapshot_id not in self._snapshots:
318 return Err(NotFoundError(f"Snapshot not found: {snapshot_id}")) # type: ignore[arg-type]
320 count = await self.data_source.restore_snapshot(snapshot_id)
321 snapshot = self._snapshots.pop(snapshot_id)
323 return Ok(
324 BulkActionResult(
325 affected_count=count,
326 metadata={
327 "action_name": snapshot.action_name,
328 "timestamp": snapshot.timestamp.isoformat(),
329 },
330 )
331 )
333 def get_recent_snapshots(self, limit: int = 10) -> list[BulkActionSnapshot]:
334 """Get recent snapshots available for undo."""
335 snapshots = sorted(
336 self._snapshots.values(),
337 key=lambda s: s.timestamp,
338 reverse=True,
339 )
340 return snapshots[:limit]
343def bulk_action(
344 name: str,
345 label: str,
346 icon: str | None = None,
347 confirm: bool = True,
348 danger: bool = False,
349) -> Any:
350 """Decorator to register a custom bulk action."""
352 def decorator(func: Callable) -> Any:
353 func._bulk_action_meta = { # type: ignore[attr-defined]
354 "name": name,
355 "label": label,
356 "icon": icon,
357 "confirm": confirm,
358 "danger": danger,
359 }
360 return func
362 return decorator