flock CLI (Python3)

Applies (or removes) an advisory lock on a specified file

Although it doesn't have feature-parity, the script is compatible with flock(1).

The script was created because, although the OS has the relevant API (i.e. flock(2)), Mac OS doesn't have the flock cli utility.

This script provides a (mostly) drop-in cross-platform option.

The script only supports acquiring exclusive locks (in blocking or non-blocking mode)

Details

Snippet

#!/usr/bin/env python3
#
# Mac OS friendly flock wrapper
#
# Author: B Tasker
#
'''
Usage:

    ./flock.py [-on] lockfile [-c] command
    ./flock.py [-un] lockfile

Note: does not support the following

    -s
    -w [timeout]

By default, if an unsupported argument is passed
the script will exit with an error (to prevent unexpected
behaviour).

However, to simply ignore these arguments, export an
environment variable:

    FLOCK_IGNORE_UNSUPPORTED=1
'''
import fcntl
import os
import sys

class FLock:
    ''' Calls the OS's flock interface
    '''

    def __init__(self):
        self.lock_f = False
        self.downstream_command = False
        self.args = False
        self.lockfile = "lock"
        self.ignore_unsupported = os.getenv("FLOCK_IGNORE_UNSUPPORTED", False)
        self.unsupported = [
            "-s",
            "--shared",
            "-w",
            "--wait",
            "--timeout"
            ]

    def acquire_lock(self):
        ''' Acquire the lock
        '''
        if not self.lock_f:
            self.open_lock_file()

        op = fcntl.LOCK_EX

        if "-n" in self.args or "--nb" in self.args:
            op |= fcntl.LOCK_NB

        r = fcntl.flock(self.lock_f, op)

    def check_command(self):
        ''' Iterate through the commandline and throw an error
        if unsupported args are provided (and not suppressed)
        '''

        if self.ignore_unsupported:
            return

        for arg in self.args:
            if arg in self.unsupported:
                print(f"Err: arg {arg} not supported")
                sys.exit(1)

    def exec(self, cmd = False):
        ''' Acquire a lock, execute the command
        then release the lock

        If cmd is not provided, the calculated command
        in self.downstream_command will be used

        '''
        self.check_command()

        if not cmd:
            cmd = ' '.join(self.downstream_command)

        if "-u" in self.args:
            # User called the unlock command
            self.open_lock_file()
            self.release()
            return

        self.acquire_lock()

        if "-o" in self.args or "--close" in self.args:
            self.lock_f.close()

        os.system(cmd)
        self.release()

    def open_lock_file(self):
        ''' Open a handle on the lockfile
        '''
        self.lock_f = open(self.lockfile, "w")

    def process_command_line(self, cmdline):
        ''' Iterate through command line arguments in order to
        identify what args (and value) have been provided to "flock"
        and what's to be passed into the command we execute
        '''

        # Process command line flags
        flock_flags = ["-s", "-x", "-e", "--exclusive", "-o", "-n", "-c", "--nb", "--close"]
        flock_flag_w_arg = ["-w"]

        accept_next = False
        flock_end = False
        last_arg = False
        lockfile = False

        built_command = []
        flock_args = {}

        for i in cmdline:
            if i == cmdline[0]:
                continue

            if i in flock_flags:
                flock_args[i] = False
                continue

            if i in flock_flag_w_arg:
                flock_args[i] = False
                accept_next = True
                last_arg = i
                continue

            if accept_next:
                flock_args[last_arg] = i
                accept_next = False
                continue

            if not lockfile:
                # We've reached the lockfile definition
                lockfile = i
                continue

            if len(built_command) == 0 and i.startswith("-"):
                # User forgot to provide the lockfile name and
                # just provided the command.
                #
                # Lets not overwrite their command...
                built_command.append(lockfile)
                lockfile = "lock"

            # We're now onto the command and its arguments
            built_command.append(i)

        self.args = flock_args
        self.downstream_command = built_command
        if lockfile:
            self.lockfile = lockfile

    def release(self):
        ''' Release the lock

        '''
        if not self.lock_f:
            self.open_lock_file()

        fcntl.flock(self.lock_f, fcntl.LOCK_UN)

if __name__ == "__main__":
    flock = FLock()
    flock.process_command_line(sys.argv)
    flock.exec()

Usage Example

./flock.py mylock ./somescript
./flock.py -n lockfile ./somescript
# Or as a Python module
import flock

f = flock.FLock()

# Define the command
f.downstream_command = ['ls', '-l']

# Set the lockfile name (it'll default to lock)
f.lockfile = "./mylockfile"

# switch to nonblocking mode
f.args = {"-n" : ""}

# Run the command
f.exec()