Monitor Python Scheduled Tasks

from apscheduler.schedulers.blocking import BlockingScheduler
import requests

def backup_database():
    check_url = "https://haspulse.io/ping/YOUR_CHECK_ID"

    # Signal start
    requests.get(f"{check_url}/start")

    try:
        # Your backup logic
        run_backup()

        # Signal success
        requests.get(check_url)
    except Exception as e:
        # Signal failure
        requests.get(f"{check_url}/fail")
        raise

scheduler = BlockingScheduler()
scheduler.add_job(backup_database, 'cron', hour=3)
scheduler.start()

from celery import Celery
import requests

app = Celery('tasks')

@app.task
def process_queue():
    check_url = "https://haspulse.io/ping/YOUR_CHECK_ID"

    try:
        # Your task logic
        result = do_work()

        # Ping on success
        requests.get(check_url)
        return result
    except Exception:
        requests.get(f"{check_url}/fail")
        raise

# In celerybeat schedule:
# 'process-queue': {
#     'task': 'tasks.process_queue',
#     'schedule': crontab(minute='*/5'),
# }

import requests
from contextlib import contextmanager

@contextmanager
def haspulse_monitor(check_id):
    """Context manager for monitoring jobs."""
    url = f"https://haspulse.io/ping/{check_id}"

    requests.get(f"{url}/start")
    try:
        yield
        requests.get(url)
    except Exception:
        requests.get(f"{url}/fail")
        raise

# Usage
with haspulse_monitor("YOUR_CHECK_ID"):
    run_my_scheduled_task()