INNER CODE UNIT · Python

_execute_job

MichielMe/fastscheduler · src/fastscheduler/main.py:455

    def _execute_job(self, job: Job):
        """Execute a job with retries."""
        if job.func is None:
            logger.error(f"Job {job.func_name} has no function, skipping")
            return

        with self.lock:
            self._running_jobs.add(job.job_id)

        job.status = JobStatus.RUNNING
        job.last_run = time.time()
        job.run_count += 1

        self._log_history(
            job.job_id,
            job.func_name,
            JobStatus.RUNNING,
            run_count=job.run_count,

View source record →