Coverage for src / lexigram / admin / services / background_jobs.py: 0%
157 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +0800
1"""Background job queue for admin import/export operations.
3Wraps import and export services to run large operations asynchronously with
4real-time progress tracking via :class:`ProgressTrackerProtocol`.
6Usage::
8 from lexigram.admin.services.background_jobs import BackgroundJobService
9 from lexigram.tasks.progress import InMemoryProgressTracker
11 tracker = InMemoryProgressTracker()
12 job_svc = BackgroundJobService(
13 progress_tracker=tracker,
14 import_service=importer,
15 export_service=exporter,
16 )
18 # Enqueue a CSV import — returns immediately with a job_id
19 job_id = await job_svc.enqueue_import(
20 resource_type="user",
21 file_content=csv_bytes,
22 file_format="csv",
23 actor_id="admin-user",
24 )
26 # Poll progress
27 status = await job_svc.get_status(job_id)
28 # → {"job_id": "...", "percent": 42, "status": "running", ...}
30 # Enqueue CSV/JSON export
31 job_id = await job_svc.enqueue_export(
32 resource_type="user",
33 export_format="csv",
34 filters={"active": True},
35 actor_id="admin-user",
36 )
37"""
39from __future__ import annotations
41import asyncio
42from dataclasses import dataclass, field
43from datetime import UTC, datetime
44from typing import Any
46from lexigram.contracts.infra.tasks.progress import ProgressTrackerProtocol
47from lexigram.logging import get_logger
48from lexigram.serialization import dumps_str
50logger = get_logger(__name__)
53class JobNotFoundError(KeyError):
54 """Raised when a job_id is not found in the queue."""
56 _code: str = "LEX_ERR_ADMIN_028"
59@dataclass
60class BackgroundJob:
61 """Runtime record of a background operation.
63 Attributes:
64 job_id: Unique identifier.
65 job_type: ``"import"`` or ``"export"``.
66 resource_type: Admin resource name.
67 actor_id: Who triggered the job.
68 status: ``"pending"``, ``"running"``, ``"completed"``, or ``"failed"``.
69 percent: 0-100 completion percentage.
70 created_at: UTC timestamp.
71 completed_at: UTC timestamp (when finished).
72 result: Final result data (on completion).
73 error: Error message (on failure).
74 metadata: Extra context (format, filename, etc.).
75 """
77 job_id: str
78 job_type: str
79 resource_type: str
80 actor_id: str
81 status: str = "pending"
82 percent: int = 0
83 created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
84 completed_at: datetime | None = None
85 result: dict[str, Any] = field(default_factory=dict)
86 error: str = ""
87 metadata: dict[str, Any] = field(default_factory=dict)
89 def to_dict(self) -> dict[str, Any]:
90 """Serialise to a JSON-safe dict for SSE/polling endpoints."""
91 return {
92 "job_id": self.job_id,
93 "job_type": self.job_type,
94 "resource_type": self.resource_type,
95 "actor_id": self.actor_id,
96 "status": self.status,
97 "percent": self.percent,
98 "created_at": self.created_at.isoformat(),
99 "completed_at": self.completed_at.isoformat()
100 if self.completed_at
101 else None,
102 "result": self.result,
103 "error": self.error,
104 "metadata": self.metadata,
105 }
108class BackgroundJobService:
109 """Manages background import/export jobs with progress tracking.
111 All heavy work runs in a background :func:`asyncio.Task`. Progress is
112 broadcast via :class:`ProgressTrackerProtocol` so subscribers can receive
113 live updates and callers can poll :meth:`get_status` without blocking.
115 Args:
116 progress_tracker: Optional :class:`ProgressTrackerProtocol` instance.
117 When provided, jobs broadcast live updates so the admin progress
118 UI works automatically.
119 import_service: Optional ``AdminImportService`` instance.
120 export_service: Optional ``ExportService`` instance.
121 max_retained_jobs: Maximum number of completed/failed jobs to keep
122 in memory before oldest are evicted.
123 """
125 def __init__(
126 self,
127 progress_tracker: ProgressTrackerProtocol | None = None,
128 import_service: Any = None,
129 export_service: Any = None,
130 max_retained_jobs: int = 200,
131 ) -> None:
132 self._progress = progress_tracker
133 self._importer = import_service
134 self._exporter = export_service
135 self._max_retained = max_retained_jobs
136 self._jobs: dict[str, BackgroundJob] = {}
137 self._tasks: dict[str, asyncio.Task[None]] = {}
138 self._background_tasks: set[asyncio.Task[None]] = set()
139 self._counter = 0
141 # ------------------------------------------------------------------
142 # Internal helpers
143 # ------------------------------------------------------------------
145 def _new_id(self, prefix: str) -> str:
146 self._counter += 1
147 ts = datetime.now(UTC).strftime("%Y%m%d%H%M%S%f")
148 return f"{prefix}-{ts}-{self._counter}"
150 def _register(self, job: BackgroundJob) -> None:
151 self._jobs[job.job_id] = job
152 self._evict_old()
154 def _evict_old(self) -> None:
155 done = [j for j in self._jobs.values() if j.status in ("completed", "failed")]
156 if len(done) > self._max_retained:
157 for old in sorted(done, key=lambda j: j.created_at)[
158 : len(done) - self._max_retained
159 ]:
160 del self._jobs[old.job_id]
162 async def _update(
163 self, job: BackgroundJob, percent: int, status: str = "running"
164 ) -> None:
165 job.percent = percent
166 job.status = status
167 if self._progress:
168 try:
169 await self._progress.update(
170 job.job_id, percent, 100, f"{status} ({percent}%)"
171 )
172 except (TimeoutError, OSError, RuntimeError):
173 pass
175 async def _finish(self, job: BackgroundJob, result: dict[str, Any]) -> None:
176 job.status = "completed"
177 job.percent = 100
178 job.result = result
179 job.completed_at = datetime.now(UTC)
180 if self._progress:
181 try:
182 await self._progress.complete(job.job_id, dumps_str(result))
183 except (TimeoutError, OSError, RuntimeError):
184 pass
186 async def _fail(self, job: BackgroundJob, error: str) -> None:
187 job.status = "failed"
188 job.error = error
189 job.completed_at = datetime.now(UTC)
190 if self._progress:
191 try:
192 await self._progress.fail(job.job_id, error)
193 except (TimeoutError, OSError, RuntimeError):
194 pass
196 # ------------------------------------------------------------------
197 # Enqueue import
198 # ------------------------------------------------------------------
200 async def enqueue_import(
201 self,
202 resource_type: str,
203 file_content: bytes,
204 *,
205 file_format: str = "csv",
206 actor_id: str = "system",
207 filename: str = "",
208 options: dict[str, Any] | None = None,
209 ) -> str:
210 """Enqueue a background import job.
212 Args:
213 resource_type: Resource name (e.g. ``"user"``).
214 file_content: Raw file bytes.
215 file_format: ``"csv"`` or ``"json"``.
216 actor_id: Principal who triggered the import.
217 filename: Optional original filename (stored in metadata).
218 options: Extra options forwarded to the import service.
220 Returns:
221 JobProtocol ID for polling via :meth:`get_status`.
222 """
223 job_id = self._new_id("import")
224 job = BackgroundJob(
225 job_id=job_id,
226 job_type="import",
227 resource_type=resource_type,
228 actor_id=actor_id,
229 metadata={"file_format": file_format, "filename": filename},
230 )
231 self._register(job)
233 task = asyncio.create_task(
234 self._run_import(job, file_content, file_format, options or {})
235 )
236 self._background_tasks.add(task)
237 task.add_done_callback(self._background_tasks.discard)
238 self._tasks[job_id] = task
239 task.add_done_callback(lambda _t: self._tasks.pop(job_id, None))
240 logger.info("Enqueued import job %s for resource '%s'", job_id, resource_type)
241 return job_id
243 async def _run_import(
244 self,
245 job: BackgroundJob,
246 file_content: bytes,
247 file_format: str,
248 options: dict[str, Any],
249 ) -> None:
250 try:
251 await self._update(job, 5)
253 if self._importer is None:
254 # Simulate when no service wired (useful for testing)
255 await asyncio.sleep(0)
256 await self._finish(
257 job, {"rows_imported": 0, "note": "no import service configured"}
258 )
259 return
261 await self._update(job, 20)
262 import_job = await self._importer.parse(
263 file_content, file_format=file_format, **options
264 )
265 await self._update(job, 60)
267 result = await self._importer.commit(import_job)
268 await self._update(job, 95)
270 if result.is_ok():
271 r = result.unwrap()
272 await self._finish(
273 job,
274 {
275 "rows_imported": getattr(r, "imported", 0),
276 "rows_skipped": getattr(r, "skipped", 0),
277 "rows_failed": getattr(r, "failed", 0),
278 },
279 )
280 else:
281 err = result.unwrap_err()
282 await self._fail(job, str(err))
284 except Exception as exc: # noqa: BLE001 — top-level task runner must capture all failures to mark job as failed
285 logger.exception("Import job %s failed: %s", job.job_id, exc)
286 await self._fail(job, str(exc))
288 # ------------------------------------------------------------------
289 # Enqueue export
290 # ------------------------------------------------------------------
292 async def enqueue_export(
293 self,
294 resource_type: str,
295 *,
296 export_format: str = "csv",
297 actor_id: str = "system",
298 filters: dict[str, Any] | None = None,
299 columns: list[str] | None = None,
300 ) -> str:
301 """Enqueue a background export job.
303 Args:
304 resource_type: Resource name.
305 export_format: ``"csv"``, ``"json"``, or ``"xlsx"``.
306 actor_id: Principal who triggered the export.
307 filters: Optional filters to apply to the query.
308 columns: Optional list of columns to include.
310 Returns:
311 JobProtocol ID for polling via :meth:`get_status`.
312 """
313 job_id = self._new_id("export")
314 job = BackgroundJob(
315 job_id=job_id,
316 job_type="export",
317 resource_type=resource_type,
318 actor_id=actor_id,
319 metadata={"export_format": export_format, "columns": columns or []},
320 )
321 self._register(job)
323 task = asyncio.create_task(
324 self._run_export(job, export_format, filters or {}, columns or [])
325 )
326 self._background_tasks.add(task)
327 task.add_done_callback(self._background_tasks.discard)
328 self._tasks[job_id] = task
329 task.add_done_callback(lambda _t: self._tasks.pop(job_id, None))
330 logger.info("Enqueued export job %s for resource '%s'", job_id, resource_type)
331 return job_id
333 async def _run_export(
334 self,
335 job: BackgroundJob,
336 export_format: str,
337 filters: dict[str, Any],
338 columns: list[str],
339 ) -> None:
340 try:
341 await self._update(job, 10)
343 if self._exporter is None:
344 await asyncio.sleep(0)
345 await self._finish(
346 job, {"rows_exported": 0, "note": "no export service configured"}
347 )
348 return
350 await self._update(job, 30)
351 output = await self._exporter.export(
352 job.resource_type,
353 format=export_format,
354 filters=filters,
355 columns=columns,
356 )
357 await self._update(job, 90)
358 await self._finish(
359 job,
360 {
361 "rows_exported": getattr(output, "row_count", 0),
362 "download_url": getattr(output, "download_url", ""),
363 "export_format": export_format,
364 },
365 )
367 except Exception as exc: # noqa: BLE001 — top-level task runner must capture all failures to mark job as failed
368 logger.exception("Export job %s failed: %s", job.job_id, exc)
369 await self._fail(job, str(exc))
371 # ------------------------------------------------------------------
372 # Status / query
373 # ------------------------------------------------------------------
375 async def get_status(self, job_id: str) -> dict[str, Any] | None:
376 """Return current job status as a JSON-safe dict.
378 Args:
379 job_id: JobProtocol identifier returned by :meth:`enqueue_import` /
380 :meth:`enqueue_export`.
382 Returns:
383 Status dict or ``None`` if job_id not found.
384 """
385 job = self._jobs.get(job_id)
386 return job.to_dict() if job else None
388 async def get_job(self, job_id: str) -> BackgroundJob | None:
389 """Return the :class:`BackgroundJob` for *job_id*, or ``None``."""
390 return self._jobs.get(job_id)
392 def list_jobs(
393 self,
394 *,
395 resource_type: str | None = None,
396 job_type: str | None = None,
397 status: str | None = None,
398 ) -> list[BackgroundJob]:
399 """Return jobs matching the given filters.
401 Args:
402 resource_type: Filter by resource type.
403 job_type: ``"import"`` or ``"export"``.
404 status: Filter by status string.
405 """
406 jobs = list(self._jobs.values())
407 if resource_type:
408 jobs = [j for j in jobs if j.resource_type == resource_type]
409 if job_type:
410 jobs = [j for j in jobs if j.job_type == job_type]
411 if status:
412 jobs = [j for j in jobs if j.status == status]
413 return sorted(jobs, key=lambda j: j.created_at, reverse=True)
415 async def cancel(self, job_id: str) -> bool:
416 """Cancel a pending or running job.
418 Args:
419 job_id: JobProtocol to cancel.
421 Returns:
422 ``True`` if the job was cancelled, ``False`` if not found or
423 already finished.
424 """
425 job = self._jobs.get(job_id)
426 if job is None or job.status in ("completed", "failed"):
427 return False
429 task = self._tasks.get(job_id)
430 if task and not task.done():
431 task.cancel()
433 job.status = "failed"
434 job.error = "Cancelled by user"
435 job.completed_at = datetime.now(UTC)
436 return True
439__all__ = [
440 "BackgroundJob",
441 "BackgroundJobService",
442 "JobNotFoundError",
443]