Portalocker

https://github.com/wolph/portalocker
Admin
Portalocker is a Python library that provides a simple and reliable way to implement file locking...
Tokens:69,106
Snippets:848
Trust Score:9.7
License:BSD-3-Clause
Update:1 month ago
Show doc for...
Tokens:
Raw
### Get Filenames Usage Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Example of retrieving and printing the list of slot files.

```python
sem = portalocker.BoundedSemaphore(3, name='pool', directory='')
print(sem.get_filenames())
# [PosixPath('pool.00.lock'), PosixPath('pool.01.lock'), PosixPath('pool.02.lock')]
```

--------------------------------

### Install Portalocker

Source: https://github.com/wolph/portalocker/blob/develop/docs/quickstart.rst

Install the base package or optional extras for Redis or Windows-specific features.

```console
pip install portalocker
```

```console
pip install "portalocker[redis]"
pip install "portalocker[win32]"
```

--------------------------------

### Install Windows dependencies

Source: https://github.com/wolph/portalocker/blob/develop/README.rst

Install the optional win32 extra for shared lock support on Windows.

```bash
pip install "portalocker[win32]"
```

--------------------------------

### Install Portalocker

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/00-index.md

Commands to install the core library and optional dependencies for Redis or Windows shared locks.

```bash
pip install portalocker              # Core
pip install "portalocker[redis]"     # With Redis
pip install "portalocker[win32]"     # With Windows shared locks
```

--------------------------------

### BoundedSemaphore Usage Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Example of initializing a BoundedSemaphore with a custom name and directory.

```python
import portalocker

# Explicitly named to avoid deprecation warning
semaphore = portalocker.BoundedSemaphore(
    maximum=3,
    name='my_worker_pool',
    directory='/tmp/locks'
)
```

--------------------------------

### Acquiring a RedisLock

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/05-redis-lock.md

Example demonstrating how to instantiate and acquire a RedisLock.

```python
import portalocker

lock = portalocker.RedisLock('job', timeout=10)
acquired = lock.acquire()  # Returns self
print(f"Lock acquired, holder_id={lock.holder_id}")
```

--------------------------------

### Acquire Slot Usage Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Example of acquiring a slot and releasing it after use.

```python
import portalocker

pool = portalocker.BoundedSemaphore(
    maximum=2,
    name='db_connections'
)

slot = pool.acquire(timeout=30)
if slot is None:
    print("No slots available")
else:
    print(f"Acquired slot: {slot.filename}")
    pool.release()
```

--------------------------------

### Verify PyPI and Clean Install

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Checks PyPI metadata and performs a smoke test installation using uv.

```bash
curl -fsSL https://pypi.org/pypi/portalocker/4.0.0/json \
  | jq -e '.info.version == "4.0.0" and (.urls | length == 2)'
pypi_smoke_dir=$(mktemp -d)
uv venv "$pypi_smoke_dir"
uv pip install --python "$pypi_smoke_dir/bin/python" \
  'portalocker==4.0.0'
"$pypi_smoke_dir/bin/python" -c \
  "import portalocker; assert portalocker.__version__ == '4.0.0'"
```

--------------------------------

### Install Redis dependencies

Source: https://github.com/wolph/portalocker/blob/develop/README.rst

Install the optional redis extra to enable Redis-based locking.

```bash
pip install "portalocker[redis]"
```

--------------------------------

### Manual Slot Sweep Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Example of using try_lock with a custom order of filenames.

```python
pool = portalocker.BoundedSemaphore(2, name='pool', directory='')
filenames = pool.get_random_filenames()  # Randomized order
if pool.try_lock(filenames):
    print(f"Got slot: {pool.lock.filename}")
    pool.release()
```

--------------------------------

### Initialize RedisLock Instances

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/05-redis-lock.md

Examples showing how to instantiate RedisLock with either an auto-created connection or a pre-existing Redis connection.

```python
import redis
import portalocker

# Auto-create connection
lock = portalocker.RedisLock('job_queue', timeout=30)

# Use existing connection
conn = redis.Redis(host='redis.local', port=6379)
lock = portalocker.RedisLock('task_lock', connection=conn, flags=portalocker.LockFlags.SHARED)
```

--------------------------------

### Smoke-test isolated wheel installation

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Verify the package installation and functionality in a clean virtual environment.

```bash
smoke_dir=$(mktemp -d)
uv venv "$smoke_dir"
uv pip install --python "$smoke_dir/bin/python" \
  dist/portalocker-4.0.0-py3-none-any.whl
"$smoke_dir/bin/python" -c \
  "import portalocker; assert portalocker.__version__ == '4.0.0'"
python dist/portalocker-4.0.0.py --help
```

--------------------------------

### Reentrant Acquisition Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/02-rlock-class.md

Demonstrates how multiple calls to acquire return the same file handle and require matching release calls.

```python
import portalocker

lock = portalocker.RLock('data.lock')
fh1 = lock.acquire()
fh2 = lock.acquire()  # Returns same fh, cheap

print(fh1 is fh2)  # True

lock.release()  # Decrements count to 1
print(fh1.closed)  # False - still acquired

lock.release()  # Count drops to 0, closes
print(fh1.closed)  # True
```

--------------------------------

### Run Redis live tests

Source: https://github.com/wolph/portalocker/blob/develop/README.rst

Commands to start a local Redis server and execute the live test suite via tox.

```bash
redis-server
tox -e redis-live
```

--------------------------------

### TemporaryFileLock.acquire Usage Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/03-temporary-pidfile-locks.md

Manual acquisition and release of a temporary lock.

```python
lock = portalocker.TemporaryFileLock('mutex.lock')
fh = lock.acquire()
lock.release()
# lock file does not exist
```

--------------------------------

### Handle LockException

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/07-exceptions.md

Example of catching a general lock failure during acquisition.

```python
import portalocker

try:
    lock = portalocker.Lock('file.txt', fail_when_locked=True)
    fh = lock.acquire()
except portalocker.LockException as exc:
    # Could be contention (then AlreadyLocked), backend failure, etc.
    print(f"Lock failed: {exc.strerror}")
    print(f"File: {exc.fh_name}")
```

--------------------------------

### PidFileLock context usage examples

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-pidfilelock-fail-closed.md

Demonstrates the default inspection-oriented context and the fail-closed context for PidFileLock.

```python
with portalocker.PidFileLock('worker.pid') as holder_pid:
    if holder_pid is None:
        run_singleton_worker()
    else:
        print(f'worker already running as PID {holder_pid}')
```

```python
try:
    with portalocker.PidFileLock('worker.pid').fail_closed():
        run_singleton_worker()
except portalocker.AlreadyLocked as exc:
    print(f'worker already running as PID {exc.holder_pid}')
```

--------------------------------

### NamedBoundedSemaphore usage examples

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.rst

Demonstrates initializing a named semaphore and verifying the generated lock file naming convention.

```pycon
>>> semaphore = NamedBoundedSemaphore(2, name='test')
>>> str(semaphore.get_filenames()[0])
'...test.00.lock'
```

```pycon
>>> semaphore = NamedBoundedSemaphore(2)
>>> 'bounded_semaphore' in str(semaphore.get_filenames()[0])
True
```

--------------------------------

### Distributed Job Queue Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/05-redis-lock.md

Demonstrates using a RedisLock to manage job processing with a timeout and exception handling for concurrent access.

```python
import portalocker
import redis

conn = redis.Redis(decode_responses=True)
job_lock = portalocker.RedisLock(
    'job_processing',
    connection=conn,
    timeout=30
)

try:
    with job_lock:
        print("Processing job...")
        # Simulate work
        import time
        time.sleep(5)
except portalocker.AlreadyLocked:
    print("Another process is already processing")
```

--------------------------------

### Verifying lock status with ensure_held

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/05-redis-lock.md

Example showing how to verify that a lock is still held during a critical section.

```python
lock = portalocker.RedisLock('critical')
with lock:
    lock.ensure_held()  # Verify still held
    # do work
```

--------------------------------

### Using New Top-Level Namespace Features

Source: https://github.com/wolph/portalocker/blob/develop/docs/migration.rst

Examples of using PidFileLock and the re-exported NamedBoundedSemaphore.

```pycon
>>> import portalocker
>>> with portalocker.PidFileLock('worker.pid') as holder_pid:
...     holder_pid is None  # None means this process is the holder
True
>>> semaphore = portalocker.NamedBoundedSemaphore(
...     2, name='migration_demo', directory=''
... )
>>> with semaphore as slot:
...     slot is not None
True
```

--------------------------------

### Add typed, spawn-safe test helpers and regression

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-issue-115-inode-waiters-plan.md

Helpers and setup for testing inode locking behavior in a multiprocessing environment. Requires portalocker and standard library multiprocessing modules.

```python
import multiprocessing.context
import multiprocessing.process
import multiprocessing.queues
import pathlib
import typing
from unittest import mock

from portalocker import LockFlags, utils

LockKind: typing.TypeAlias = typing.Literal['temporary', 'pid']


def make_temporary_lock(
    filename: str,
    lock_kind: LockKind,
    *,
    timeout: float,
    fail_when_locked: bool,
) -> portalocker.TemporaryFileLock:
    """Create one of the temporary-path lock implementations under test."""
    lock_type: type[portalocker.TemporaryFileLock]
    if lock_kind == 'temporary':
        lock_type = portalocker.TemporaryFileLock
    else:
        lock_type = portalocker.PidFileLock
    return lock_type(
        filename,
        timeout=timeout,
        fail_when_locked=fail_when_locked,
    )


def native_lock_path(filename: str, lock_kind: LockKind) -> str:
    """Return the pathname carrying the native OS lock."""
    if lock_kind == 'pid':
        return f'{filename}.lock'
    return filename


def hold_inode_waiter(
    filename: str,
    lock_kind: LockKind,
    native_filename: str,
    opened_event: multiprocessing.synchronize.Event,
    acquired_event: multiprocessing.synchronize.Event,
    release_event: multiprocessing.synchronize.Event,
    inode_queue: multiprocessing.queues.Queue[int],
) -> None:
    """Open the original inode, acquire, report it, and hold the lock."""
    original_get_fh: typing.Callable[
        [utils.Lock],
        typing.IO[typing.Any],
    ] = typing.cast(
        typing.Callable[[utils.Lock], typing.IO[typing.Any]],
        utils.Lock._get_fh,
    )

    def announce_open(lock: utils.Lock) -> typing.IO[typing.Any]:
        fh: typing.IO[typing.Any] = original_get_fh(lock)
        if lock.filename == native_filename:
            opened_event.set()
        return fh

    lock: portalocker.TemporaryFileLock = make_temporary_lock(
        filename,
        lock_kind,
        timeout=30,
        fail_when_locked=False,
    )
    try:
        with mock.patch.object(utils.Lock, '_get_fh', announce_open):
            fh: typing.IO[typing.Any] = lock.acquire()
        inode_queue.put(os.fstat(fh.fileno()).st_ino)
        acquired_event.set()
        if not release_event.wait(timeout=30):
            raise TimeoutError('parent did not release the inode waiter')
    finally:
        lock.release()
```

--------------------------------

### Refresh and validate source refs

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Fetch remote branches and verify the state of the repository before starting the integration.

```bash
git fetch --prune origin master develop
git status --short --branch
git rev-parse origin/master origin/develop
git ls-remote --heads --tags origin master develop \
  'release/4.0.0-integration' 'v4.0.0'
```

--------------------------------

### TemporaryFileLock Usage Example

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/03-temporary-pidfile-locks.md

Basic usage of TemporaryFileLock using a context manager to ensure automatic file removal.

```python
import portalocker

lock = portalocker.TemporaryFileLock('worker.lock', timeout=10)
with lock:
    print('doing work')
# File is automatically removed on exit
```

--------------------------------

### Get all slot filenames

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.rst

Demonstrates how to retrieve a list of all available semaphore slot paths in numerical order.

```pycon
>>> semaphore = BoundedSemaphore(2, name='example', directory='')
>>> [str(filename) for filename in semaphore.get_filenames()]
['example.00.lock', 'example.01.lock']
```

--------------------------------

### Pin to Older Python Version

Source: https://github.com/wolph/portalocker/blob/develop/docs/migration.rst

Use this command to install a version compatible with Python interpreters older than 3.10.

```console
pip install "portalocker<4"
```

--------------------------------

### Get a single slot filename

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.rst

Demonstrates how to generate the path for a specific semaphore slot using get_filename.

```pycon
>>> semaphore = BoundedSemaphore(2, name='example', directory='')
>>> str(semaphore.get_filename(1))
'example.01.lock'
```

--------------------------------

### RLock Context Manager Usage

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/02-rlock-class.md

Examples of using RLock as a context manager, including nested usage and true reentrancy by sharing the instance.

```python
with portalocker.RLock('shared.lock') as fh:
    # Nested context works
    with portalocker.RLock('shared.lock') as fh2:
        fh2.write('data')
    # fh still open
```

```python
lock = portalocker.RLock('file.lock')

with lock:
    with lock:  # Same instance, truly reentrant
        pass
```

--------------------------------

### Static Analysis and Build Commands

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-open-atomic-no-replace.md

Commands for type checking, documentation generation, and distribution building.

```bash
uv run mypy
uv run basedpyright
uv run pyrefly check
uv run ty check
```

```bash
uv run codespell
mkdir -p docs/_static
uv run sphinx-build -W -b html docs docs/_build/html
```

```bash
uv build
```

--------------------------------

### Initialize RedisLock

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.redis.rst

Demonstrates basic initialization of a RedisLock instance.

```pycon
>>> from portalocker import redis
>>> lock = redis.RedisLock('some_channel')
>>> lock.client_name == f'some_channel-lock-{lock.holder_id}'
True
```

--------------------------------

### Run the combine command

Source: https://github.com/wolph/portalocker/blob/develop/docs/cli.rst

Executes the combine subcommand to bundle the package into the default output location.

```console
$ python -m portalocker combine
```

--------------------------------

### Documentation build command

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-pidfilelock-fail-closed.md

Command to build documentation using tox.

```bash
uv run tox -e docs
```

--------------------------------

### Use Timeout Generator

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/08-types-and-configuration.md

Example usage of the timeout generator for retry logic.

```python
for attempt in lock._timeout_generator(timeout=5, check_interval=0.25):
    try:
        # Try to acquire
        break
    except AlreadyLocked:
        pass  # Generator sleeps and yields next attempt number
```

--------------------------------

### Basic RedisLock usage

Source: https://github.com/wolph/portalocker/blob/develop/docs/redis.rst

Demonstrates initializing a connection and using RedisLock as a context manager.

```pycon
>>> import fakeredis
>>> import portalocker
>>> connection = fakeredis.FakeStrictRedis(
...     server=fakeredis.FakeServer(), decode_responses=True
... )
>>> with portalocker.RedisLock('some_channel', connection=connection):
...     print('do something here')
do something here
```

--------------------------------

### Handle LockLostError

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/07-exceptions.md

Example of catching LockLostError during a lock context manager block.

```python
import portalocker

lock = portalocker.RedisLock('resource')
try:
    with lock:
        # do work
        pass
except portalocker.LockLostError as exc:
    print(f"Lost lock on {exc.channel}")
    print(f"Cause: {exc.__cause__}")
    # Handle gracefully; resource may be held by new lock owner
```

--------------------------------

### Run Full Test Suite

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-issue-115-inode-waiters-plan.md

Execute the complete test suite to ensure system stability.

```bash
uv run pytest
```

--------------------------------

### Retry lock acquisition

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/07-exceptions.md

Example of handling contention by updating the timeout and retrying.

```python
import portalocker
import time

lock = portalocker.Lock('data.txt', timeout=1)
try:
    fh = lock.acquire()
except portalocker.AlreadyLocked:
    print("Lock busy, retrying with longer timeout...")
    lock.timeout = 30
    fh = lock.acquire()  # Now retries for 30 seconds
```

--------------------------------

### Build distributions

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-pidfilelock-fail-closed.md

Command to build source and wheel distributions.

```bash
uv build
```

--------------------------------

### Inspect lock holder

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/07-exceptions.md

Example of accessing the holder_pid attribute when a lock is already held.

```python
import portalocker

lock = portalocker.Lock('critical.lock', fail_when_locked=True)
try:
    fh = lock.acquire()
except portalocker.AlreadyLocked as exc:
    if exc.holder_pid:
        print(f"Lock held by PID {exc.holder_pid}")
    else:
        print(f"Lock held by another process")
```

--------------------------------

### Using TemporaryFileLock

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.rst

Demonstrates the lifecycle of a temporary lock file, including acquisition, verification, and release.

```pycon
>>> import os
>>> import portalocker
>>> lock = portalocker.TemporaryFileLock('somefile.lock')
>>> _ = lock.acquire()
>>> os.path.isfile('somefile.lock')
True
>>> lock.release()
>>> os.path.isfile('somefile.lock')
False
>>> lock.release()
```

--------------------------------

### Get filename for a specific slot

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Returns the formatted filename for a given slot number.

```python
def get_filename(self, number: int) -> pathlib.Path:
```

```python
sem = portalocker.BoundedSemaphore(2, name='example', directory='')
print(sem.get_filename(0))  # example.00.lock
print(sem.get_filename(1))  # example.01.lock
```

--------------------------------

### Configure RedisLock with Custom Settings

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/08-types-and-configuration.md

Shows how to merge custom Redis connection settings with the library defaults.

```python
import portalocker

# Merge custom settings with defaults
redis_kwargs = dict(
    host='redis.example.com',
    port=6380,
    # health_check_interval defaults to 10
)
lock = portalocker.RedisLock(
    'resource',
    redis_kwargs=redis_kwargs,
    thread_sleep_time=0.05,  # Faster polling
)
```

--------------------------------

### Get Filenames Method Definition

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Method definition for retrieving all slot file paths.

```python
def get_filenames(self) -> typing.Sequence[pathlib.Path]:
```

--------------------------------

### Using RedisLock with fakeredis

Source: https://github.com/wolph/portalocker/blob/develop/docs/lock-types.rst

Demonstrates initializing a RedisLock with a fakeredis connection and using it as a context manager.

```pycon
>>> import fakeredis
>>> import portalocker
>>> connection = fakeredis.FakeStrictRedis(
...     server=fakeredis.FakeServer(), decode_responses=True
... )
>>> with portalocker.RedisLock('example_channel', connection=connection):
...     print('do something here')
do something here
```

--------------------------------

### Display project directory structure

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/README.md

Visual representation of the portalocker package file hierarchy.

```text
portalocker/
├── exceptions.py      (exception classes)
├── constants.py       (LockFlags enum, platform constants)
├── types.py           (type aliases and protocols)
├── portalocker.py     (low-level lock/unlock functions, lockers)
├── utils.py           (Lock, RLock, TemporaryFileLock, PidFileLock, BoundedSemaphore, NamedBoundedSemaphore, open_atomic)
├── redis.py           (RedisLock)
└── __init__.py        (public API exports)
```

--------------------------------

### Proactive lock check

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/07-exceptions.md

Example of using ensure_held() to verify lock status during long-running operations.

```python
lock = portalocker.RedisLock('critical_section')
with lock:
    for i in range(10):
        # Check every iteration
        try:
            lock.ensure_held()
        except portalocker.LockLostError:
            print("Lock lost, aborting work")
            break
        # do work iteration
```

--------------------------------

### Get random filenames for BoundedSemaphore

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Retrieves paths for all slots in random order to help distribute contention.

```python
def get_random_filenames(self) -> typing.Sequence[pathlib.Path]:
```

```python
sem = portalocker.BoundedSemaphore(5, name='jobs', directory='')
random_order = sem.get_random_filenames()
# Each call returns a different shuffle
```

--------------------------------

### Perform release preflight checks

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Verify repository state, authentication, and environment readiness before executing the release.

```bash
git fetch --prune origin master develop
git status --porcelain
git rev-parse HEAD origin/develop origin/master
git merge-base --is-ancestor origin/master HEAD
git ls-remote --tags origin v4.0.0
gh release view v4.0.0 --repo wolph/portalocker
curl -fsSL https://pypi.org/pypi/portalocker/json \
  | jq -e '.releases["4.0.0"] | length == 0'
key_id=$(git config --get user.signingkey)
gpg --batch --list-secret-keys "$key_id"
gh auth status
release_helper=$(command -v build_and_upload_release)
test -n "$release_helper"
zsh -n "$release_helper"
```

--------------------------------

### Implement Singleton Service with PidFileLock

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/09-usage-patterns.md

Uses a PID file to ensure only one instance of a service runs at a time.

```python
import portalocker
import sys

def run_service():
    with portalocker.PidFileLock('app.pid') as holder_pid:
        if holder_pid is None:
            print("Starting service...")
            # do work
        else:
            print(f"Service already running as PID {holder_pid}")
            sys.exit(1)
```

--------------------------------

### Test with Lock

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/00-index.md

Demonstrates a basic unit test verifying file creation while using a lock.

```python
def test_with_lock():
    with portalocker.Lock('test.txt', 'w') as fh:
        fh.write('test')
    assert os.path.exists('test.txt')
```

--------------------------------

### Run project verification

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-release-error-policy.md

Commands to run full test suite and environment checks.

```bash
uv run pytest
uv run tox -p auto
```

--------------------------------

### BoundedSemaphore.__init__

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Initializes a new BoundedSemaphore instance with a specified number of slots and configuration for lock file management.

```APIDOC
## BoundedSemaphore.__init__

### Description
Initializes a cross-process counting semaphore that limits concurrent holders to a maximum number of slots.

### Parameters
- **maximum** (int) - Required - Number of slots (concurrent holders allowed)
- **name** (str) - Optional - Shared name identifying this semaphore
- **filename_pattern** (str) - Optional - Template for slot filenames
- **directory** (str) - Optional - Directory holding slot files
- **timeout** (float | None) - Optional - Retry timeout
- **check_interval** (float | None) - Optional - Retry interval
- **fail_when_locked** (bool | None) - Optional - If True, raise AlreadyLocked when all slots busy after timeout
```

--------------------------------

### Restore implementation

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-issue-115-inode-waiters-plan.md

The correct implementation of _acquire_verified that includes inode verification.

```python
        for _ in lock._timeout_generator(timeout, check_interval):
            fh = Lock.acquire(lock, timeout, check_interval, fail_when_locked)
            if os.name == 'nt':  # Windows: a locked file can't be swapped.
                return fh  # pragma: not-nt
            if _fh_matches_path(fh, filename):  # pragma: not-posix
                return fh  # pragma: not-posix
            # Stale handle: the path was unlinked+recreated behind our back.
            Lock.release(lock)  # pragma: not-posix
        raise exceptions.AlreadyLocked(  # pragma: not-posix
            exceptions.LockException.LOCK_FAILED,
            f'{filename!r} kept being replaced while locking (split-brain)',
        )
```

--------------------------------

### Build and validate release artifacts

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Generate distribution files and verify them using Twine.

```bash
rm -rf build dist
uv build
uv run build_extra.py
uvx twine check dist/portalocker-4.0.0-py3-none-any.whl \
  dist/portalocker-4.0.0.tar.gz
ls -l dist/portalocker-4.0.0-py3-none-any.whl \
  dist/portalocker-4.0.0.tar.gz dist/portalocker-4.0.0.py
```

--------------------------------

### Run Static Analysis Checks

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-issue-115-inode-waiters-plan.md

Execute all configured static analysis tools to verify code quality.

```bash
uv run ruff check
uv run ruff format --check
uv run mypy
uv run basedpyright
uv run pyrefly check
uv run ty check
uv run codespell
```

--------------------------------

### acquire(timeout=None, check_interval=None, fail_when_locked=None)

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.utils.rst

Opens the file, acquires the lock, and returns the filehandle.

```APIDOC
## acquire(timeout=None, check_interval=None, fail_when_locked=None)

### Description
Opens the file, locks it, and returns the filehandle. If the lock is already held, the existing filehandle is returned.

### Parameters
- **timeout** (float | None) - Optional - Overrides the instance timeout.
- **check_interval** (float | None) - Optional - Overrides the instance check_interval.
- **fail_when_locked** (bool | None) - Optional - Overrides the instance fail_when_locked setting.

### Returns
- **IO[Any]** - The open, locked filehandle.

### Raises
- **AlreadyLocked** - Raised if the file is locked and fail_when_locked is True or timeout expires.
- **LockException** - Raised if the backend fails for reasons other than contention.
- **OSError** - Raised if file opening or preparation fails.
```

--------------------------------

### Initialize RedisLockHolder

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.redis.rst

Demonstrates the initialization of a RedisLockHolder instance and accessing its properties.

```pycon
>>> from portalocker import redis
>>> holder = redis.RedisLockHolder('a1b2', redis.RedisLockMode.SHARED)
>>> holder.holder_id, holder.mode.value, holder.legacy
('a1b2', 'shared', False)
>>> holder.elected is None
True
```

--------------------------------

### Run focused and core locking tests

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-release-error-policy.md

Command to verify the implementation against core locking and release error tests.

```bash
uv run pytest \
  portalocker_tests/test_release_errors.py \
  portalocker_tests/test_core_locking.py \
  portalocker_tests/test_rlock_behaviour.py \
  portalocker_tests/test_temporary_file_lock.py \
  -q --no-cov
```

--------------------------------

### Run test and static-analysis gates

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Execute the test suite and static analysis tools to ensure code quality.

```bash
uv sync --all-extras
uv run pytest
uv run tox -e ruff,mypy,basedpyright,pyrefly,ty,codespell,repo-review,docs
uv run pyright
```

--------------------------------

### Pin a locker class or use an instance

Source: https://github.com/wolph/portalocker/blob/develop/docs/platforms.rst

Demonstrates assigning a specific locker class to the global default or using an instance directly to avoid affecting global state.

```python
import portalocker.portalocker
from portalocker.portalocker import LockfLocker

portalocker.portalocker.LOCKER = LockfLocker

# Or use one directly, leaving the global default alone:
locker = LockfLocker()
with open('data.txt', 'w') as fh:
    locker.lock(fh, portalocker.LockFlags.EXCLUSIVE)
    locker.unlock(fh)
```

--------------------------------

### Initialize PidFileLock

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/03-temporary-pidfile-locks.md

Basic instantiation of a PidFileLock object.

```python
import portalocker

lock = portalocker.PidFileLock('daemon.pid')
# At this point, no files created yet
```

--------------------------------

### Acquire a Redis lock

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.redis.rst

Demonstrates initializing a RedisLock with a fake connection and acquiring it with a timeout.

```pycon
>>> import fakeredis
>>> import portalocker
>>> connection = fakeredis.FakeStrictRedis(
...     server=fakeredis.FakeServer(), decode_responses=True
... )
>>> lock = portalocker.RedisLock(
...     'some_channel', connection=connection
... )
>>> lock.acquire(timeout=1) is lock
True
>>> lock.release()
```

--------------------------------

### Instantiate and inspect LockLostError

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.exceptions.rst

Demonstrates creating a LockLostError instance and accessing its specific attributes.

```pycon
>>> from portalocker import exceptions
>>> error = exceptions.LockLostError(
...     exceptions.LockException.LOCK_FAILED,
...     'lock lost',
...     channel='jobs',
...     holder_id='a1b2',
... )
>>> error.channel, error.holder_id
('jobs', 'a1b2')
```

--------------------------------

### Generate BoundedSemaphore filenames

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.rst

Demonstrates how to initialize a BoundedSemaphore and retrieve a list of potential lock filenames.

```pycon
>>> semaphore = BoundedSemaphore(2, name='example', directory='')
>>> names = semaphore.get_random_filenames()
>>> sorted(str(filename) for filename in names)
['example.00.lock', 'example.01.lock']
```

--------------------------------

### Initialize RedisLock with and without external connections

Source: https://github.com/wolph/portalocker/blob/develop/docs/redis.rst

Demonstrates how RedisLock behaves when a connection is provided versus when it is initialized standalone.

```pycon
>>> given = portalocker.RedisLock('given_channel', connection=connection)
>>> given.connection is connection
True
>>> standalone = portalocker.RedisLock('standalone_channel')
>>> standalone.connection is None
True
```

--------------------------------

### NamedBoundedSemaphore Usage

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/04-bounded-semaphore.md

Demonstrates initializing the semaphore with an explicit name or an auto-generated one.

```python
import portalocker

# Named explicitly
sem = portalocker.NamedBoundedSemaphore(
    maximum=5,
    name='worker_pool'
)

# Auto-generated name (for single-process use only)
sem = portalocker.NamedBoundedSemaphore(maximum=2)
```

--------------------------------

### Access package metadata

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/08-types-and-configuration.md

Retrieve and print package version and author information.

```python
import portalocker

print(f"portalocker {portalocker.__version__}")
print(f"Author: {portalocker.__author__}")
```

--------------------------------

### Execute release helper

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-08-01-portalocker-4.0.0-release.md

Run the release helper script using the extracted release message.

```bash
"$release_helper" "$release_message"
```

--------------------------------

### Lock a file using portalocker

Source: https://github.com/wolph/portalocker/blob/develop/docs/api/portalocker.utils.rst

Demonstrates the basic usage of portalocker.Lock as a context manager to safely lock a file.

```pycon
>>> import portalocker
>>> with portalocker.Lock('somefile', 'w', timeout=1) as fh:
...     _ = fh.write('locked while the block runs')
```

--------------------------------

### Implement a custom locker

Source: https://github.com/wolph/portalocker/blob/develop/docs/platforms.rst

Shows how to create a custom locker by subclassing BaseLocker and overriding the lock and unlock methods.

```pycon
>>> import portalocker
>>> from portalocker.portalocker import BaseLocker
>>> class RecordingLocker(BaseLocker):
...     def lock(self, file_obj, flags):
...         print('lock', portalocker.LockFlags(flags).name)
...     def unlock(self, file_obj):
...         print('unlock')
>>> previous = portalocker.portalocker.LOCKER
>>> portalocker.portalocker.LOCKER = RecordingLocker
>>> try:
...     with open('example.txt', 'w') as fh:
...         portalocker.lock(fh, portalocker.LockFlags.EXCLUSIVE)
...         portalocker.unlock(fh)
... finally:
...     portalocker.portalocker.LOCKER = previous
lock EXCLUSIVE
unlock
```

--------------------------------

### Generic LockBase Usage

Source: https://github.com/wolph/portalocker/blob/develop/docs/migration.rst

Demonstrates the generic typing of LockBase and how to implement a custom subclass.

```pycon
>>> import typing
>>> from portalocker.utils import LockBase
>>> LockBase[typing.IO[typing.Any]]
portalocker.utils.LockBase[typing.IO[typing.Any]]
>>> class CountingLock(LockBase):
...     def acquire(
...         self, timeout=None, check_interval=None, fail_when_locked=None
...     ):
...         return 42
...
...     def release(self):
...         pass
>>> CountingLock().acquire()
42
```

--------------------------------

### Simulate pre-fix behavior

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-issue-115-inode-waiters-plan.md

Temporary replacement for _acquire_verified to simulate the bug where split-inode race conditions were not detected.

```python
        return Lock.acquire(
            lock,
            timeout,
            check_interval,
            fail_when_locked,
        )
```

--------------------------------

### Inspect POSIX LockException structure

Source: https://github.com/wolph/portalocker/blob/develop/docs/troubleshooting.rst

Demonstrates the structure of a LockException on POSIX systems using a simulated BlockingIOError.

```pycon
>>> from portalocker import exceptions
>>> original = BlockingIOError(11, 'Resource temporarily unavailable')
>>> exc = exceptions.LockException(original, str(original))
>>> str(exc)
"(BlockingIOError(11, 'Resource temporarily unavailable'), '[Errno 11] Resource temporarily unavailable')"
>>> exc.strerror
'[Errno 11] Resource temporarily unavailable'
```

--------------------------------

### Use a BoundedSemaphore

Source: https://github.com/wolph/portalocker/blob/develop/docs/lock-types.rst

Shows basic usage of a BoundedSemaphore with a specified name and directory.

```pycon
>>> import portalocker
>>> semaphore = portalocker.BoundedSemaphore(2, name='workers', directory='')
>>> with semaphore as lock:
...     lock is not None
True
```

--------------------------------

### Specify output file for combine

Source: https://github.com/wolph/portalocker/blob/develop/docs/cli.rst

Uses the --output-file or -o flag to define a custom destination path for the generated vendorable file.

```console
$ python -m portalocker combine --output-file vendor/portalocker.py
$ python -m portalocker combine -o vendor/portalocker.py
```

--------------------------------

### Use Default Lock Configuration

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/08-types-and-configuration.md

Demonstrates that calling the Lock constructor without arguments uses the module-level defaults.

```python
import portalocker

# These are identical:
lock1 = portalocker.Lock('file.txt')
lock2 = portalocker.Lock(
    'file.txt',
    timeout=5,
    check_interval=0.25,
    fail_when_locked=False,
    flags=portalocker.LOCK_METHOD,
)
```

--------------------------------

### Acquire and release a TemporaryFileLock

Source: https://github.com/wolph/portalocker/blob/develop/docs/lock-types.rst

Demonstrates the lifecycle of a TemporaryFileLock, showing that the lock file is created upon acquisition and removed upon release.

```pycon
>>> import os
>>> import portalocker
>>> lock = portalocker.TemporaryFileLock('example.lock')
>>> _ = lock.acquire()
>>> os.path.isfile('example.lock')
True
>>> lock.release()
>>> os.path.isfile('example.lock')
False
```

--------------------------------

### Constructor for BaseLockException

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/07-exceptions.md

Initializes the exception with optional filehandle information and arguments.

```python
def __init__(
    self,
    *args: typing.Any,
    fh: types.IO | int | types.HasFileno | None = None,
    **kwargs: typing.Any,
) -> None:
```

--------------------------------

### Run linting and type checking

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-issue-115-inode-waiters-plan.md

Commands to verify code quality and type safety of the test file.

```bash
uv run ruff check portalocker_tests/test_multiprocess.py
uv run ruff format --check portalocker_tests/test_multiprocess.py
uv run mypy portalocker_tests/test_multiprocess.py
```

--------------------------------

### TemporaryFileLock.__init__

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/03-temporary-pidfile-locks.md

Initializes a new TemporaryFileLock instance, which registers with the interpreter-exit cleanup hook to ensure the lock file is removed.

```APIDOC
## TemporaryFileLock.__init__

### Description
Initializes a new TemporaryFileLock instance. This lock variant automatically removes the lock file upon release.

### Parameters
- **filename** (str | pathlib.Path) - Optional - Path to the temporary lock file (default: '.lock')
- **timeout** (float | None) - Optional - Retry timeout (default: None)
- **check_interval** (float) - Optional - Retry interval (default: 0.25)
- **fail_when_locked** (bool) - Optional - Whether to fail fast when locked (default: True)
- **flags** (LockFlags) - Optional - Lock flags (default: EXCLUSIVE | NON_BLOCKING)
- **raise_on_release_error** (bool) - Optional - Whether to report file unlink errors instead of logging (default: False)
```

--------------------------------

### Test Execution Commands

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-open-atomic-no-replace.md

Commands to run focused tests and verify the implementation.

```bash
uv run pytest --no-cov portalocker_tests/test_open_atomic.py -v
```

```bash
uv run pytest --no-cov portalocker_tests/test_open_atomic.py portalocker/utils.py -v
```

```bash
uv run pytest
```

--------------------------------

### Initialize RLock

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/02-rlock-class.md

Basic instantiation of an RLock object.

```python
import portalocker

rlock = portalocker.RLock('worker.lock', timeout=5)
```

--------------------------------

### Acquire RedisLock with Fencing Tokens

Source: https://github.com/wolph/portalocker/blob/develop/docs/redis.rst

Demonstrates initializing RedisLock with fencing enabled and observing the incrementing fence token.

```pycon
>>> writer = portalocker.RedisLock(
...     'fenced_channel', connection=connection, fencing=True
... )
>>> rival = portalocker.RedisLock(
...     'fenced_channel', connection=connection, fencing=True
... )
>>> with writer:
...     writer.fence_token
1
>>> with rival:
...     rival.fence_token
2
>>> writer.fence_token  # survives release, until the next acquire
1
>>> connection.get(writer.fence_key)  # release leaves the key behind
'2'
```

--------------------------------

### Run Regression Tests

Source: https://github.com/wolph/portalocker/blob/develop/docs/superpowers/plans/2026-07-16-open-atomic-no-replace.md

Command to run the full test suite for portalocker_tests.

```bash
uv run pytest --no-cov portalocker_tests/test_open_atomic.py -v
```

--------------------------------

### Apply lock to file object

Source: https://github.com/wolph/portalocker/blob/develop/_autodocs/06-low-level-functions.md

Demonstrates locking a standard Python file object using portalocker.lock.

```python
import portalocker

fh = open('data.txt', 'r+')
portalocker.lock(fh, portalocker.LockFlags.SHARED)
# Read while shared lock held
portalocker.unlock(fh)
fh.close()
```

--------------------------------

### Initialize NamedBoundedSemaphore

Source: https://github.com/wolph/portalocker/blob/develop/docs/lock-types.rst

Demonstrates creating a named semaphore to limit concurrent processes.

```pycon
>>> import portalocker
>>> semaphore = portalocker.NamedBoundedSemaphore(2, name='workers', directory='')
>>> with semaphore as lock:
...     lock is not None
True
```
Tokens:
Raw