Runners¶

def stop(self, execution: CommandExecution) -> None:
    """Request a graceful stop of a RUNNING *execution*.

    The base implementation sets the cooperative ``stop_requested``
    flag (checked on the worker's output-flush heartbeat); backends
    with a way to signal the worker process override this to also
    send a SIGTERM-equivalent.
    """
    from django_admin_runner.models import CommandExecution

    CommandExecution.objects.filter(
        pk=execution.pk, status=CommandExecution.Status.RUNNING
    ).update(stop_requested=True)

def force_stop(self, execution: CommandExecution) -> bool:
    """Hard-kill the process running *execution*.

    Returns ``True`` when a kill was performed. The base
    implementation is a no-op (unsupported backend); runners whose
    backend runs each task in its own process override this.
    """
    return False

def finalize_stale(self, execution: CommandExecution) -> tuple[str, str] | None:
    """Attribute why a dead worker's execution ended (best effort).

    Called by the stale-run sweeper for RUNNING executions whose
    ``worker_pid`` no longer exists. Returns ``(status, note)`` —
    e.g. ``(TIMEOUT, "Killed by django-q2 timeout")`` — or ``None``
    when the backend has nothing to say (the sweeper then marks the
    execution FAILED with a generic "worker lost" note). Must never
    raise; wrap backend lookups defensively.
    """
    return None

def create_schedule(self, schedule: ScheduledCommand) -> str:
    """Materialize *schedule* into a native backend object.

    Returns the backend reference to store in
    ``ScheduledCommand.backend_schedule_key``.
    """
    self._check_schedule_kind(schedule)
    raise ScheduleNotSupportedError(
        f"Backend {self.backend or type(self).__name__!r} does not support schedules."
    )

def update_schedule(self, schedule: ScheduledCommand) -> str:
    """Update the native object referenced by ``backend_schedule_key`` in place.

    Returns the (possibly new) backend reference.
    """
    self._check_schedule_kind(schedule)
    raise ScheduleNotSupportedError(
        f"Backend {self.backend or type(self).__name__!r} does not support schedules."
    )

def delete_schedule(self, schedule: ScheduledCommand) -> None:
    """Remove the native object referenced by ``backend_schedule_key``."""
    self._check_schedule_kind(schedule)
    raise ScheduleNotSupportedError(
        f"Backend {self.backend or type(self).__name__!r} does not support schedules."
    )

def schedule_next_run(self, schedule: ScheduledCommand):
    """Next run datetime of the native object, or ``None`` when unknown."""
    return None

def stop(self, execution) -> None:
    """Graceful stop: stop flag (heartbeat) + revoke with SIGTERM."""
    super().stop(execution)
    self._revoke(execution, "SIGTERM")

def force_stop(self, execution) -> bool:
    """Hard kill: revoke with SIGKILL."""
    if not execution.task_id:
        return False
    self._revoke(execution, "SIGKILL")
    return True

def finalize_stale(self, execution):
    """Attribute a swept execution via the Celery result backend.

    REVOKED tasks (e.g. killed by a hard ``time_limit`` revoke) are
    reported as TIMEOUT; FAILURE as FAILED. Best effort — an
    unreachable broker yields ``None`` (generic FAILED).
    """
    if not execution.task_id:
        return None
    try:
        from celery.result import AsyncResult

        state = AsyncResult(execution.task_id).state
    except Exception:  # noqa: BLE001 - attribution must never break the sweep
        return None
    from django_admin_runner.models import CommandExecution

    if state == "REVOKED":
        return (
            CommandExecution.Status.TIMEOUT,
            "Revoked/killed by Celery (hard time limit or revoke).",
        )
    if state == "FAILURE":
        return (
            CommandExecution.Status.FAILED,
            "Worker died; Celery recorded a task failure.",
        )
    return None

def stop(self, execution) -> None:
    """Graceful stop: stop flag (heartbeat) + SIGTERM to the worker PID."""
    super().stop(execution)
    import signal

    self._signal_worker(execution, signal.SIGTERM)

def force_stop(self, execution) -> bool:
    """Hard kill: SIGKILL to the (verified) worker PID."""
    import signal

    return self._signal_worker(execution, signal.SIGKILL)

def finalize_stale(self, execution):
    """Attribute a swept execution via django-q2's Task records.

    The q2 sentinel writes a failure record for tasks it killed
    (e.g. for exceeding ``Q_CLUSTER["timeout"]``). Best effort — a
    missing/unreachable record yields ``None`` (generic FAILED).
    """
    if not execution.task_id:
        return None
    from django_admin_runner.models import CommandExecution

    try:
        from django_q.models import Task

        task = Task.objects.filter(name=execution.task_id).first()
    except Exception:  # noqa: BLE001 - attribution must never break the sweep
        return None
    if task is None or task.success or not task.stopped:
        return None
    result = str(task.result or "")
    if "time" in result.lower():
        return (
            CommandExecution.Status.TIMEOUT,
            f"Killed by django-q2 timeout: {result}",
        )
    return (
        CommandExecution.Status.FAILED,
        f"Worker killed; django-q2 recorded failure: {result}",
    )