CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-portalocker

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

—

Pending
Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

SecuritybySnyk

Pending

The risk profile of this skill

Overview
Eval results
Files

Portalocker

A cross-platform file locking library that provides reliable file locking mechanisms across Windows, Linux, Unix, and macOS systems. Portalocker offers both traditional file-based locks using native system calls and advanced Redis-based distributed locks for multi-process and multi-machine coordination.

Package Information

  • Package Name: portalocker
  • Language: Python
  • Installation: pip install portalocker
  • Optional Dependencies: pip install portalocker[redis] for Redis-based locking

Core Imports

import portalocker

Common usage patterns:

from portalocker import Lock, lock, unlock, LOCK_EX, LOCK_SH, LOCK_NB

For Redis-based locking:

from portalocker import RedisLock

Basic Usage

import portalocker

# Basic file locking with context manager
with portalocker.Lock('my_file.txt', 'r+') as fh:
    data = fh.read()
    # File is automatically locked here
    fh.write('new data')
    # File is automatically unlocked when exiting context

# Manual locking and unlocking
with open('my_file.txt', 'r+') as fh:
    portalocker.lock(fh, portalocker.LOCK_EX)
    try:
        # Do work with locked file
        data = fh.read()
        fh.write('modified data')
    finally:
        portalocker.unlock(fh)

# Non-blocking lock with timeout
try:
    with portalocker.Lock('my_file.txt', timeout=5.0, fail_when_locked=True) as fh:
        # Work with file
        pass
except portalocker.AlreadyLocked:
    print("File is locked by another process")

Architecture

Portalocker provides a layered architecture for cross-platform file locking:

  • High-level Lock Classes: Context managers (Lock, RLock, TemporaryFileLock) with timeout and error handling
  • Low-level Functions: Direct lock/unlock functions (lock, unlock) for manual control
  • Platform Abstraction: Automatic selection between Windows (Win32/msvcrt) and POSIX (fcntl) implementations
  • Advanced Features: Redis-based distributed locks, bounded semaphores, atomic file operations
  • Type System: Complete type definitions for all file handles, flags, and configuration options

This design ensures reliable cross-platform operation while providing flexibility from simple context managers to complex distributed locking scenarios.

Capabilities

File Locking

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.

def lock(file: FileArgument, flags: LockFlags) -> None: ...
def unlock(file: FileArgument) -> None: ...

LOCK_EX: LockFlags  # Exclusive lock
LOCK_SH: LockFlags  # Shared lock  
LOCK_NB: LockFlags  # Non-blocking
LOCK_UN: LockFlags  # Unlock

File Locking

Lock Classes

High-level lock managers with built-in timeout, context manager support, and advanced features like reentrant locks and temporary file locks.

class Lock:
    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: ...
    def release(self) -> None: ...

class RLock(Lock): ...  # Reentrant lock
class TemporaryFileLock(Lock): ...  # Auto-deleting temporary lock

Lock Classes

Redis Distributed Locking

Redis pubsub-based distributed locks that provide immediate unlocking when connections are lost, suitable for multi-process and multi-machine coordination.

class RedisLock:
    def __init__(self, channel: str, connection: redis.Redis | None = None,
                 timeout: float | None = None, check_interval: float | None = None,
                 fail_when_locked: bool | None = False, thread_sleep_time: float = 0.1,
                 unavailable_timeout: float = 1, redis_kwargs: dict | None = None) -> None: ...
    def acquire(self, timeout: float | None = None, check_interval: float | None = None,
                fail_when_locked: bool | None = None) -> 'RedisLock': ...
    def release(self) -> None: ...

Redis Locking

Semaphores and Resource Management

Bounded semaphores for limiting concurrent processes accessing shared resources, with support for named semaphores across process boundaries.

class BoundedSemaphore:  # Deprecated
    def __init__(self, maximum: int, name: str = 'bounded_semaphore', 
                 filename_pattern: str = '{name}.{number:02d}.lock',
                 directory: str = tempfile.gettempdir(), **kwargs) -> None: ...

class NamedBoundedSemaphore(BoundedSemaphore):
    def __init__(self, maximum: int, name: str | None = None, **kwargs) -> None: ...

Semaphores

Utility Functions

Additional utilities for atomic file operations and helper functions.

def open_atomic(filename: Filename, binary: bool = True) -> typing.Iterator[typing.IO]: ...

Utilities

Exception Handling

class LockException(Exception):
    """Base exception for locking errors"""

class AlreadyLocked(LockException):
    """Raised when file is already locked by another process"""

Type Definitions

from typing import Union, Literal
import pathlib

# File path types
Filename = Union[str, pathlib.Path]

# File handle types  
FileArgument = Union[typing.IO, int, HasFileno]

# File open modes
Mode = Literal['r', 'w', 'a', 'x', 'rb', 'wb', 'ab', 'xb', 'r+', 'w+', 'a+', 'x+', ...]

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

# Protocol for objects with fileno() method
class HasFileno(typing.Protocol):
    def fileno(self) -> int: ...
Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/portalocker@3.2.x
Publish Source
CLI
Badge
tessl/pypi-portalocker badge