INNER CODE UNIT · Python
_register_function
MichielMe/fastscheduler · src/fastscheduler/main.py:227
def _register_function(self, func: Callable):
"""Register a function for persistence."""
self.job_registry[f"{func.__module__}.{func.__name__}"] = func
def _next_job_id(self) -> str:
"""Generate next job ID (thread-safe)."""
self._job_counter_value = next(self._job_counter)
return f"job_{self._job_counter_value}"
def _add_job(self, job: Job):
"""Add job to the priority queue."""
with self.lock:
if any(j.job_id == job.job_id for j in self.jobs):
logger.warning(f"Job {job.job_id} already exists, skipping")
return
heapq.heappush(self.jobs, job)
self._log_history(job.job_id, job.func_name, JobStatus.SCHEDULED)