How to use the locket.lock_file function in locket

To help you get started, we’ve selected a few locket examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def lock_is_released_by_context_manager_exit(lock_path):
    has_run = False
    
    # Keep a reference to first_lock so it holds onto the lock
    first_lock = locket.lock_file(lock_path, timeout=0)
    
    with first_lock:
        pass
        
    with locket.lock_file(lock_path, timeout=0):
        has_run = True
    
    assert has_run
github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def lock_can_be_acquired_with_timeout_of_zero(lock_path):
    has_run = False
    with locket.lock_file(lock_path, timeout=0):
        has_run = True
    
    assert has_run
github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def thread_cannot_obtain_lock_using_same_path_twice_without_release(lock_path):
    with locket.lock_file(lock_path, timeout=0):
        lock = locket.lock_file(lock_path, timeout=0)
        try:
            lock.acquire()
            assert False, "Expected LockError"
        except locket.LockError:
            pass
github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def the_same_lock_file_object_is_used_for_the_same_path_with_different_arguments(lock_path):
    # We explicitly check the same lock is used to ensure that the lock isn't
    # re-entrant, even if the underlying platform lock is re-entrant.
    first_lock = locket.lock_file(lock_path, timeout=None)
    second_lock = locket.lock_file(lock_path, timeout=0)
    assert first_lock._lock is second_lock._lock
github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def different_file_objects_are_used_for_different_paths(lock_path):
    first_lock = locket.lock_file(lock_path, timeout=0)
    second_lock = locket.lock_file(lock_path + "-2", timeout=0)
    assert first_lock._lock is not second_lock._lock
github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def the_same_lock_file_object_is_used_for_the_same_path_with_different_arguments(lock_path):
    # We explicitly check the same lock is used to ensure that the lock isn't
    # re-entrant, even if the underlying platform lock is re-entrant.
    first_lock = locket.lock_file(lock_path, timeout=None)
    second_lock = locket.lock_file(lock_path, timeout=0)
    assert first_lock._lock is second_lock._lock
github mwilliamson / locket.py / tests / locket_tests.py View on Github external
def different_file_objects_are_used_for_different_paths(lock_path):
    first_lock = locket.lock_file(lock_path, timeout=0)
    second_lock = locket.lock_file(lock_path + "-2", timeout=0)
    assert first_lock._lock is not second_lock._lock
github kizniche / Mycodo / mycodo / controller_input.py View on Github external
def read_adc(self):
        """ Read voltage from ADC """
        try:
            lock_acquired = False

            # Set up lock for ADC
            adc_lock = locket.lock_file(self.adc_lock_file, timeout=120)
            try:
                adc_lock.acquire()
                lock_acquired = True
            except:
                self.logger.error("Could not acquire ADC lock. Breaking for future locking.")
                os.remove(self.adc_lock_file)

            if not lock_acquired:
                self.logger.error(
                    "Unable to acquire lock: {lock}".format(
                        lock=self.adc_lock_file))

            # Get measurement from ADC
            measurements = self.adc.next()

            if measurements is not None:

locket

File-based locks for Python on Linux and Windows

BSD-2-Clause
Latest version published 3 years ago

Package Health Score

68 / 100
Full package analysis

Similar packages