python process lock

README.md

Simple and lightweight python process LOCK

Use a lock across python processes or threads.

Based on sqlite/sqlalchemy implementation. Tested on python 3.4+.

To see how it works launch process.py multiple times.

notes

• if your process terminates without unlocking the lock is removed by sqlite, since the connection is garbaged.

• to install the log package used in the example:

$ pip3 install rapydo-utils==0.5.1

process.py

# -*- coding: utf-8 -*-
# /usr/bin/env python3

import time
from sqlock import Lock
from utilities.logs import get_logger

log = get_logger(__name__)

lock = Lock()

if __name__ == '__main__':

    log.debug("Process starting")

    lock.acquire()
    try:
        time.sleep(30)
    except KeyboardInterrupt:
        pass
    lock.unlock()

    log.info("Process completed")

sqlock.py

# -*- coding: utf-8 -*-

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# to install this package: $ pip3 install rapydo-utils
from utilities.logs import get_logger

log = get_logger(__name__)
Base = declarative_base()


class LockTable(Base):
    __tablename__ = 'lock'
    id = Column(Integer, primary_key=True)


class Lock(object):

    def __init__(self):
        super(Lock, self).__init__()
        self.dbfile = 'restapi.lock'
        self.get_session()

    def get_session(self):

        engine = create_engine(
            'sqlite:///%s' % self.dbfile,
            connect_args={'timeout': 3}
        )

        try:
            Base.metadata.create_all(engine)
        except BaseException:
            pass

        Base.metadata.bind = engine
        DBSession = sessionmaker(bind=engine)
        self.session = DBSession()
        return self.session

    def acquire(self):
        if self.check_locked():
            log.exit('Lock seems already acquired')
        else:
            self.session.execute('BEGIN EXCLUSIVE TRANSACTION;')
            log.verbose('acquired lock')

    def unlock(self):
        self.session.execute('COMMIT TRANSACTION;')

    def check_locked(self):
        try:
            self.session.query(LockTable).all()
        except BaseException:
            return True
        else:
            return False