A global named lock. An easy way to prevent multiple copies of a process from running concurrently

socklock.py

import sys
import socket


class SocketLock:
    """ A global named lock. Better than pidfiles.

    This uses a bind() on AF_UNIX sockets as a means to acquire a lock that is non-blocking.
    Socket files are not created in the real filesystem, but in an "abstract namespace":
    a Linux-only feature of hidden socket files.

    This way of locking is ideal because it's released as soon as your application quits.
    But it does not work in Windows. Don't use Windows.
    And it does not work on Mac. It's your fault in this case.
    """
    __slots__ = ('socket_name', '_sock', 'is_acquired')

    # Only works in Linux
    is_supported = (sys.platform == 'linux')

    def __init__(self, path: str):
        """

        Args:
             path: Arbitrary path in the abstract socket namespace.
                It's global! Transcends chroots!
        """
        self.socket_name = path
        self.is_acquired = False

    def acquire(self) -> bool:
        """ Attempt to acquire the lock

        Note that you can do it only once, and then you'll need to release() it.
        The lock will be released automatically if your process quits.

        Returns:
            True/False whether the lock has been acquired
        """
        # Not supported? Do nothing
        if not self.is_supported:
            return True

        # Already acquired?
        assert not self._sock, 'Already acquired'

        # Bind to an abstract socket
        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        try:
            # The null byte (\0) means the the socket is created in the abstract namespace. Linux only
            self._sock.bind('\0' + self.socket_name)
        # Result?
        except socket.error:
            return False
        else:
            self.is_acquired = True
            return True

    def release(self):
        """ Release the lock """
        if not self.is_supported:
            return

        assert self._sock, 'Was never acquired'

        self._sock.close()
        self._sock = None
        self.is_acquired = False