pg_try_advisory_lock snippets

How to use PostgreSQL advisory locks with SQLAlchemy and Python

Tagged python, advisory lock, pg_try_advisory_lock, sqlalchemy  Languages python

advisory_lock.py

from models import Session
from sqlalchemy import func, select


def execute(session, lock_fn, lock_id, scope):
    """
    Executes the lock function
    """
    return session.execute(select([lock_fn(lock_id, scope)])).scalar()


def obtain_lock(session, lock_id, scope):
    """
    Obtains the advisory lock
    """
    lock_fn = func.pg_try_advisory_lock
    return execute(session, lock_fn, lock_id, scope)


def release_lock(session, lock_id, scope):
    """
    Releases the advisory lock
    """
    lock_fn = func.pg_advisory_unlock
    return execute(session, lock_fn, lock_id, scope)


def with_lock(my_func, lock_id, scope=1):
    """
    Executes my_func if the lock can be obtained.
    """
    session = Session()
    obtained_lock = False
    try:
        obtained_lock = obtain_lock(session, lock_id, scope)
        if obtained_lock:
            my_func()
    finally:
        if obtained_lock:
            release_lock(session, lock_id, scope)

Usage:

from advisory_lock import with_lock

def run():
    print("It runs")

if __name__ == '__main__':
    with_lock(run, 300000)