Running periodic jobs once with Celery beat and PostgreSQL advisory locks

# ======================================================================
# Q: What scheduler do you use for periodic jobs and how do you avoid double
# execution?
#
# Celery beat (or APScheduler, or a Kubernetes CronJob). The scheduler runs on a
# single instance, but for safety (a deploy with two pods for a few seconds) I
# put a lock inside the job. The PostgreSQL advisory lock is a good choice if I
# don't want Redis:
# Each job records start/end/status in a job_runs table, so there is an alert if
# a job "did not run" (dead man's switch), not only if it failed.
# ======================================================================

from sqlalchemy import text

def run_exclusive(session, job_key: int, fn):
    got = session.scalar(text('SELECT pg_try_advisory_xact_lock(:k)'), {'k': job_key})
    if not got:
        return 'skipped: already running'
    return fn(session)        # the lock is released automatically at the end of the transaction