tessl/pypi-portalocker

Cross-platform file locking library that provides reliable file locking mechanisms across Windows, Linux, Unix, and macOS systems

—

Pending

Overview

Eval results

Files

File Locking

Name: tessl/pypi-portalocker
Author: tessl

Core file locking functionality using advisory locks with support for exclusive/shared locks, non-blocking mode, and timeout handling. Works across Windows, Linux, Unix, and macOS with automatic platform-specific implementation selection.

Capabilities

Basic Locking Functions

Low-level file locking functions that provide direct control over file locks using native system calls.

def lock(file: FileArgument, flags: LockFlags) -> None:
    """
    Lock a file using advisory locking.
    
    Parameters:
    - file: File object, file descriptor, or object with fileno() method
    - flags: Locking flags (LOCK_EX, LOCK_SH, LOCK_NB combinations)
    
    Raises:
    - LockException: If locking fails due to system error
    - AlreadyLocked: If file is already locked and LOCK_NB is specified
    """

def unlock(file: FileArgument) -> None:
    """
    Unlock a previously locked file.
    
    Parameters:
    - file: File object, file descriptor, or object with fileno() method
    
    Raises:
    - LockException: If unlocking fails due to system error
    """

Lock Constants

Flag constants for controlling lock behavior, automatically set to appropriate values for the current platform.

LOCK_EX: LockFlags  # Exclusive lock - only one process can hold this lock
LOCK_SH: LockFlags  # Shared lock - multiple processes can hold shared locks simultaneously  
LOCK_NB: LockFlags  # Non-blocking - fail immediately if lock cannot be acquired
LOCK_UN: LockFlags  # Unlock - remove existing lock (used internally)

# Enum version with the same values
class LockFlags(enum.IntFlag):
    EXCLUSIVE: int     # Same as LOCK_EX
    SHARED: int        # Same as LOCK_SH  
    NON_BLOCKING: int  # Same as LOCK_NB
    UNBLOCK: int       # Same as LOCK_UN

Usage Examples

Basic file locking with manual lock/unlock:

import portalocker

# Exclusive lock (default)
with open('data.txt', 'r+') as fh:
    portalocker.lock(fh, portalocker.LOCK_EX)
    try:
        # File is now exclusively locked
        data = fh.read()
        fh.seek(0)
        fh.write('modified: ' + data)
        fh.truncate()
    finally:
        portalocker.unlock(fh)

Non-blocking lock with error handling:

import portalocker

try:
    with open('data.txt', 'r+') as fh:
        # Try to lock immediately, fail if already locked
        portalocker.lock(fh, portalocker.LOCK_EX | portalocker.LOCK_NB)
        # Work with file
        data = fh.read()
        portalocker.unlock(fh)
except portalocker.AlreadyLocked:
    print("File is currently locked by another process")
except portalocker.LockException as e:
    print(f"Locking failed: {e}")

Shared locks for read-only access:

import portalocker

# Multiple processes can hold shared locks simultaneously
with open('config.txt', 'r') as fh:
    portalocker.lock(fh, portalocker.LOCK_SH)
    try:
        config_data = fh.read()
        # Process read-only data
    finally:
        portalocker.unlock(fh)

Working with file descriptors:

import os
import portalocker

# Lock using file descriptor
fd = os.open('data.txt', os.O_RDWR)
try:
    portalocker.lock(fd, portalocker.LOCK_EX)
    # Work with file descriptor
    data = os.read(fd, 1024)
    os.write(fd, b'new data')
    portalocker.unlock(fd)
finally:
    os.close(fd)

Platform-Specific Behavior

Windows Implementation

Uses Win32 API (LockFileEx/UnlockFileEx) or msvcrt.locking
Supports both exclusive and shared locks
File position is preserved during locking operations
Automatic fallback between implementation methods

POSIX Implementation

Uses fcntl.flock for advisory locking
Supports exclusive, shared, and non-blocking modes
Works with file descriptors and file objects
Compatible with Linux, Unix, macOS, and other POSIX systems

Error Handling

Common error scenarios and their exceptions:

# File already locked by another process
try:
    portalocker.lock(fh, portalocker.LOCK_EX | portalocker.LOCK_NB)
except portalocker.AlreadyLocked as e:
    print(f"File locked by: {e.fh}")

# System-level locking error
try:
    portalocker.lock(fh, portalocker.LOCK_EX)
except portalocker.LockException as e:
    print(f"Locking failed: {e.strerror}")

Type Definitions

from typing import Union, Protocol
import typing
import io

# File argument types accepted by lock/unlock functions
FileArgument = Union[typing.IO[typing.Any], io.TextIOWrapper, int, HasFileno]

class HasFileno(Protocol):
    """Protocol for objects that have a fileno() method"""
    def fileno(self) -> int: ...

# Lock flags enum
class LockFlags(enum.IntFlag):
    EXCLUSIVE: int     # Exclusive lock
    SHARED: int        # Shared lock
    NON_BLOCKING: int  # Non-blocking mode
    UNBLOCK: int       # Unlock operation

Install with Tessl CLI