Lock Classes

class Lock:
    """
    Lock manager with built-in timeout and context manager support.
    
    Parameters:
    - filename: Path to file to lock (str or pathlib.Path)
    - mode: File open mode ('a', 'r+', 'w+', etc.). 'w' modes truncate after lock acquisition
    - timeout: Timeout in seconds when trying to acquire lock (default: 5.0) 
    - check_interval: Check interval while waiting for lock (default: 0.25)
    - fail_when_locked: Fail immediately if initial lock fails (default: False)
    - flags: Lock flags (default: LOCK_EX | LOCK_NB)
    - **file_open_kwargs: Additional arguments passed to open()
    """
    
    def __init__(self, filename: Filename, mode: str = 'a', timeout: float | None = None,
                 check_interval: float = 0.25, fail_when_locked: bool = False, 
                 flags: LockFlags = LOCK_EX | LOCK_NB, **file_open_kwargs) -> None: ...
    
    def acquire(self, timeout: float | None = None, check_interval: float | None = None,
                fail_when_locked: bool | None = None) -> typing.IO:
        """
        Acquire the file lock and return file handle.
        
        Parameters:
        - timeout: Override default timeout
        - check_interval: Override default check interval  
        - fail_when_locked: Override default fail_when_locked behavior
        
        Returns:
        - File handle for the locked file
        
        Raises:
        - AlreadyLocked: If lock cannot be acquired and fail_when_locked=True
        - LockException: If locking fails due to system error
        """
    
    def release(self) -> None:
        """Release the currently held lock and close file handle"""
    
    def __enter__(self) -> typing.IO:
        """Context manager entry - acquire lock and return file handle"""
    
    def __exit__(self, exc_type, exc_value, traceback) -> None:
        """Context manager exit - release lock"""

class RLock(Lock):
    """
    Reentrant lock that can be acquired multiple times by the same process.
    Must be released the same number of times it was acquired.
    """
    
    def __init__(self, filename: Filename, mode: str = 'a', timeout: float = 5.0,
                 check_interval: float = 0.25, fail_when_locked: bool = False,
                 flags: LockFlags = LOCK_EX | LOCK_NB) -> None: ...
    
    def acquire(self, timeout: float | None = None, check_interval: float | None = None,
                fail_when_locked: bool | None = None) -> typing.IO:
        """Acquire lock (can be called multiple times by same process)"""
    
    def release(self) -> None:
        """Release lock (must match number of acquire() calls)"""

class TemporaryFileLock(Lock):
    """
    Temporary file lock that auto-deletes the lock file on release.
    Automatically registers cleanup with atexit.
    """
    
    def __init__(self, filename: str = '.lock', timeout: float = 5.0,
                 check_interval: float = 0.25, fail_when_locked: bool = True,
                 flags: LockFlags = LOCK_EX | LOCK_NB) -> None: ...
    
    def release(self) -> None:
        """Release lock and delete the temporary lock file"""

import portalocker

# Simple file locking with automatic cleanup
with portalocker.Lock('data.txt', 'r+', timeout=10.0) as fh:
    # File is automatically locked here
    data = fh.read()
    fh.seek(0)
    fh.write('modified: ' + data)
    fh.truncate()
# File is automatically unlocked and closed here

import portalocker

# Create lock object
lock = portalocker.Lock('data.txt', mode='r+', timeout=5.0)

try:
    # Acquire lock
    fh = lock.acquire()
    
    # Work with file
    data = fh.read()
    fh.write('new data')
    
finally:
    # Always release lock
    lock.release()

import portalocker

def process_file_nested():
    with portalocker.RLock('data.txt', 'r+') as fh1:
        # First lock acquisition
        data = fh1.read()
        
        # Nested function that also needs the same lock
        with portalocker.RLock('data.txt', 'r+') as fh2:
            # Second lock acquisition by same process - succeeds
            fh2.write('nested access: ' + data)
        
        # First lock still held here
        fh1.write('outer access completed')
    # All locks released here

import portalocker

# Create a temporary lock for process coordination
with portalocker.TemporaryFileLock('/tmp/my_process.lock') as fh:
    # Only one instance of this process can run
    print("Starting exclusive process...")
    do_exclusive_work()
    print("Process completed")
# Lock file is automatically deleted

import portalocker

try:
    # Fail immediately if file is already locked
    with portalocker.Lock('data.txt', fail_when_locked=True, timeout=0) as fh:
        process_file(fh)
except portalocker.AlreadyLocked:
    print("File is currently being processed by another instance")

import portalocker

# Pass additional parameters to open()
with portalocker.Lock('data.txt', 'r+', encoding='utf-8', buffering=1) as fh:
    # File opened with custom encoding and line buffering
    text_data = fh.read()
    fh.write('unicode data: ' + text_data)

import portalocker
import time

# Custom timeout and check intervals
lock = portalocker.Lock(
    'data.txt',
    timeout=30.0,        # Wait up to 30 seconds
    check_interval=0.5,  # Check every 500ms
    fail_when_locked=False  # Keep retrying until timeout
)

try:
    with lock:
        # Will retry acquiring lock for up to 30 seconds
        process_file()
except portalocker.LockException:
    print("Could not acquire lock within 30 seconds")

try:
    with portalocker.Lock('data.txt', fail_when_locked=True) as fh:
        process_file(fh)
except portalocker.AlreadyLocked as e:
    print(f"File is locked: {e}")
except portalocker.LockException as e:
    print(f"Lock failed: {e}")
except FileNotFoundError:
    print("File does not exist")

from typing import Union
import pathlib
import typing

Filename = Union[str, pathlib.Path]

class LockFlags(enum.IntFlag):
    EXCLUSIVE: int
    SHARED: int
    NON_BLOCKING: int
    UNBLOCK: int

# Default lock method used by Lock classes
LOCK_METHOD = LockFlags.EXCLUSIVE | LockFlags.NON_BLOCKING