A thread-safe manager for creating, updating, and retrieving information about background jobs. It stores job information in memory.
| 64 | return remaining_files * avg_time_per_file |
| 65 | |
| 66 | class JobManager: |
| 67 | """ |
| 68 | A thread-safe manager for creating, updating, and retrieving information |
| 69 | about background jobs. It stores job information in memory. |
| 70 | """ |
| 71 | def __init__(self): |
| 72 | self.jobs: Dict[str, JobInfo] = {} |
| 73 | self.lock = threading.Lock() # A lock to ensure thread-safe access to the jobs dictionary. |
| 74 | |
| 75 | def create_job(self, path: str, is_dependency: bool = False) -> str: |
| 76 | """Creates a new job, assigns it a unique ID, and stores it.""" |
| 77 | job_id = str(uuid.uuid4()) |
| 78 | with self.lock: |
| 79 | self.jobs[job_id] = JobInfo( |
| 80 | job_id=job_id, |
| 81 | status=JobStatus.PENDING, |
| 82 | start_time=datetime.now(), |
| 83 | path=path, |
| 84 | is_dependency=is_dependency |
| 85 | ) |
| 86 | return job_id |
| 87 | |
| 88 | def update_job(self, job_id: str, **kwargs): |
| 89 | """Updates the information for a specific job in a thread-safe manner.""" |
| 90 | with self.lock: |
| 91 | if job_id in self.jobs: |
| 92 | job = self.jobs[job_id] |
| 93 | for key, value in kwargs.items(): |
| 94 | if hasattr(job, key): |
| 95 | setattr(job, key, value) |
| 96 | |
| 97 | def get_job(self, job_id: str) -> Optional[JobInfo]: |
| 98 | """Retrieves the information for a single job.""" |
| 99 | with self.lock: |
| 100 | return self.jobs.get(job_id) |
| 101 | |
| 102 | def list_jobs(self) -> List[JobInfo]: |
| 103 | """Returns a list of all jobs currently in the manager.""" |
| 104 | with self.lock: |
| 105 | return list(self.jobs.values()) |
| 106 | |
| 107 | def find_active_job_by_path(self, path: str) -> Optional[JobInfo]: |
| 108 | """Finds the most recent, currently active (pending or running) job for a given path.""" |
| 109 | with self.lock: |
| 110 | path_obj = Path(path).resolve() |
| 111 | |
| 112 | matching_jobs = sorted( |
| 113 | [job for job in self.jobs.values() if job.path and Path(job.path).resolve() == path_obj], |
| 114 | key=lambda j: j.start_time, |
| 115 | reverse=True |
| 116 | ) |
| 117 | |
| 118 | for job in matching_jobs: |
| 119 | if job.status in [JobStatus.PENDING, JobStatus.RUNNING]: |
| 120 | return job |
| 121 | |
| 122 | return None |
| 123 |
no outgoing calls