Clean up old job status entries.
(self)
| 286 | return job_id.replace('--', '/') |
| 287 | |
| 288 | def cleanup_old_jobs(self): |
| 289 | """Clean up old job status entries.""" |
| 290 | cutoff = datetime.now() - timedelta(hours=WebAppConfig.JOB_CLEANUP_HOURS) |
| 291 | all_jobs = self.background_worker.get_all_jobs() |
| 292 | expired_jobs = [ |
| 293 | job_id for job_id, job in all_jobs.items() |
| 294 | if job.created_at < cutoff and job.status in ['completed', 'failed'] |
| 295 | ] |
| 296 | |
| 297 | for job_id in expired_jobs: |
| 298 | if job_id in self.background_worker.job_status: |
| 299 | del self.background_worker.job_status[job_id] |
no test coverage detected