File size: 2,508 Bytes
63c75d5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import time
from datetime import datetime, timezone, timedelta
from typing import Optional
from spooler.store import get_conn
def claim_lock(job_name: str, owner: str, ttl_minutes: int = 60) -> bool:
"""Attempt to claim a distributed lock for a job name. Returns True if claimed."""
conn = get_conn()
now_iso = datetime.now(timezone.utc).isoformat()
until_iso = (datetime.now(timezone.utc) + timedelta(minutes=ttl_minutes)).isoformat()
# Check if a valid lock exists
row = conn.execute(
"SELECT locked_until FROM scheduler_jobs WHERE job_name = ?",
(job_name,)
).fetchone()
if row and row["locked_until"] and row["locked_until"] > now_iso:
# Lock is currently held and not expired
conn.close()
return False
# Claim the lock (insert if new, overwrite if expired)
conn.execute(
"""
INSERT OR REPLACE INTO scheduler_jobs
(job_name, last_run_at, locked_until, locked_by, status)
VALUES (?, COALESCE((SELECT last_run_at FROM scheduler_jobs WHERE job_name = ?), ?), ?, ?, 'locked')
""",
(job_name, job_name, now_iso, until_iso, owner)
)
conn.commit()
conn.close()
return True
def release_lock(job_name: str, owner: str, status_msg: str = "completed") -> bool:
"""Release a held lock, providing a status message."""
conn = get_conn()
now_iso = datetime.now(timezone.utc).isoformat()
# Ensure this owner actually holds the lock
row = conn.execute(
"SELECT locked_by FROM scheduler_jobs WHERE job_name = ?",
(job_name,)
).fetchone()
if not row or row["locked_by"] != owner:
conn.close()
return False
conn.execute(
"""
UPDATE scheduler_jobs
SET locked_until = NULL,
locked_by = NULL,
last_run_at = ?,
status = ?
WHERE job_name = ? AND locked_by = ?
""",
(now_iso, status_msg, job_name, owner)
)
conn.commit()
conn.close()
return True
def get_job_status(job_name: Optional[str] = None) -> list[dict]:
"""Get status of all jobs or a specific job."""
conn = get_conn()
if job_name:
rows = conn.execute(
"SELECT * FROM scheduler_jobs WHERE job_name = ?",
(job_name,)
).fetchall()
else:
rows = conn.execute("SELECT * FROM scheduler_jobs").fetchall()
conn.close()
return [dict(row) for row in rows]
|