A data class to hold all information about a single background job. This makes it easy to track the job's progress, status, and results.
| 22 | |
| 23 | @dataclass |
| 24 | class JobInfo: |
| 25 | """ |
| 26 | A data class to hold all information about a single background job. |
| 27 | This makes it easy to track the job's progress, status, and results. |
| 28 | """ |
| 29 | job_id: str |
| 30 | status: JobStatus |
| 31 | start_time: datetime |
| 32 | end_time: Optional[datetime] = None |
| 33 | total_files: int = 0 |
| 34 | processed_files: int = 0 |
| 35 | current_file: Optional[str] = None |
| 36 | status_message: Optional[str] = None |
| 37 | estimated_duration: Optional[float] = None |
| 38 | actual_duration: Optional[float] = None |
| 39 | errors: List[str] = None |
| 40 | result: Optional[Dict[str, Any]] = None |
| 41 | path: Optional[str] = None |
| 42 | is_dependency: bool = False |
| 43 | |
| 44 | def __post_init__(self): |
| 45 | """Ensures the errors list is initialized after the object is created.""" |
| 46 | if self.errors is None: |
| 47 | self.errors = [] |
| 48 | |
| 49 | @property |
| 50 | def progress_percentage(self) -> float: |
| 51 | """Calculates the completion percentage of the job.""" |
| 52 | if self.total_files == 0: |
| 53 | return 0.0 |
| 54 | return (self.processed_files / self.total_files) * 100 |
| 55 | |
| 56 | @property |
| 57 | def estimated_time_remaining(self) -> Optional[float]: |
| 58 | """Calculates the estimated time remaining based on the average time per file.""" |
| 59 | if self.status != JobStatus.RUNNING or self.processed_files == 0: |
| 60 | return None |
| 61 | elapsed = (datetime.now() - self.start_time).total_seconds() |
| 62 | avg_time_per_file = elapsed / self.processed_files |
| 63 | remaining_files = self.total_files - self.processed_files |
| 64 | return remaining_files * avg_time_per_file |
| 65 | |
| 66 | class JobManager: |
| 67 | """ |