filelock

https://py-filelock.readthedocs.io/en/latest/
Admin
filelock is a platform-independent file locking library for Python that provides inter-process...
Tokens:128,609
Snippets:882
Trust Score:9.6
Update:4 weeks ago
Show doc for...
Tokens:
Raw
### Install filelock via pip

Source: https://py-filelock.readthedocs.io/en/latest/_sources/index.rst.txt

Install the filelock package from PyPI using pip.

```bash
python -m pip install filelock
```

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

### Start tracked release

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/asyncio.html

Starts the release process with fork tracking, choosing between async and sync release methods based on whether the release method is a coroutine function.

```python
    def _start_tracked_release(self) -> asyncio.Future[_BackendOutcome[None]]:
        return self._start_internal_method(
            self._release_with_fork_tracking_async
            if iscoroutinefunction(self._release)
            else self._release_with_fork_tracking
        )
```

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

### Inspect protocol 2 owner record with SoftFileLease

Source: https://py-filelock.readthedocs.io/en/latest/how-to.html

Creates a SoftFileLease and reads the owner record, which includes pid, hostname, mode, token, lease_duration, and start. The owner is None if the marker is missing, malformed, or protocol 1.

```python
from filelock import SoftFileLease

lease = SoftFileLease("work.lock", lease_duration=30)

if (owner := lease.owner) is None:
    print("no marker, or its record is malformed or protocol 1")
else:
    print(owner.pid, owner.hostname)
    print(owner.mode)            # "lease", or "unknown" for a contract this version cannot interpret
    print(owner.token)           # claim identity
    print(owner.lease_duration)  # 30.0
    print(owner.start)           # process start token, or None where unavailable
```

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

### Read lease owner record (pid, hostname, mode, token, lease_duration, start)

Source: https://py-filelock.readthedocs.io/en/latest/_sources/how-to.rst.txt

Inspect the current lease owner's record. `owner` reads the marker on disk each time, so it reports whoever currently holds the path, not necessarily this process. `None` is ambiguous: a missing marker, a malformed one, and a protocol 1 one all read the same. `owner.mode` reads `"unknown"` when the marker names a contract this version does not implement; treat that as "held by something I should not touch".

```python
if (owner := lease.owner) is None:
    print("no marker, or its record is malformed or protocol 1")
else:
    print(owner.pid, owner.hostname)
    print(owner.mode)            # "lease", or "unknown" for a contract this version cannot interpret
    print(owner.token)           # claim identity
    print(owner.lease_duration)  # 30.0
    print(owner.start)           # process start token, or None where unavailable
```

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

### Basic FileLock usage

Source: https://py-filelock.readthedocs.io/en/latest/_sources/index.rst.txt

Shows how to use FileLock to synchronize access to a file. The lock is acquired using a context manager, and the file is opened and written within the lock.

```python
from filelock import FileLock

lock = FileLock("high_ground.txt.lock")
with lock:
    with open("high_ground.txt", "a") as f:
        f.write("You were the chosen one.")
```

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

### FileLock.__init__

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_api.html

Documentation for the FileLock constructor, which initializes a new lock object with configurable options.

```APIDOC
## FileLock.__init__

### Description
Creates a new lock object with configurable timeout, mode, thread-local context, blocking behavior, and other options.

### Method
Constructor

### Signature
`__init__(self, lock_file: str | os.PathLike[str], timeout: float = -1, mode: int = _UNSET_FILE_MODE, thread_local: bool = True, *, blocking: bool = True, is_singleton: bool = False, poll_interval: float = 0.05, lifetime: float | None = None, context_error_policy: ContextErrorPolicy = "chain", close_error_policy: CloseErrorPolicy = "default", fallback_to_soft: bool = True, preserve_lock_file: bool = False, on_acquired: Callable[[int], None] | None = None) -> None`

### Parameters
- **lock_file** (str | os.PathLike[str]) - Required - Path to the file.
- **timeout** (float) - Optional - Default timeout when acquiring the lock, in seconds. Used as fallback in acquire method if no timeout value (None) is given. Negative value disables timeout; 0 means exactly one attempt.
- **mode** (int) - Optional - File permissions for the lockfile. When not specified, OS controls permissions via umask and default ACLs.
- **thread_local** (bool) - Optional - Whether internal context is thread local. If False, lock is reentrant across threads. When True (default), all fields of internal context are per-thread.
- **blocking** (bool) - Optional - Whether the lock should be blocking or not.
- **is_singleton** (bool) - Optional - If True, only one instance of this class will be created per lock file.
- **poll_interval** (float) - Optional - Default interval for polling the lock file, in seconds. Used as fallback in acquire method if no poll_interval value (None) is given.
- **lifetime** (float | None) - Optional - For SoftFileLock, age in seconds after which a waiting process may delete the marker. None disables age-based expiry. Native OS locks ignore non-None lifetime with a warning.
- **context_error_policy** (ContextErrorPolicy) - Optional - How a context manager reconciles a failure in its body with a failure while releasing on exit. "chain" (default) keeps Python's behavior; "group" raises a BaseExceptionGroup.
- **close_error_policy** (CloseErrorPolicy) - Optional - What to do with an os.close failure after relinquishing descriptor ownership. "default" keeps backend's historical behavior; "raise" always propagates OSError; "suppress" always ignores it.
- **fallback_to_soft** (bool) - Optional - For UnixFileLock, whether to switch to SoftFileLock when flock returns ENOSYS. True (default) keeps historical fallback; False fails closed.
- **preserve_lock_file** (bool) - Optional - For native locks, whether filelock promises not to unlink the lock pathname on release. False (default) keeps each backend's cleanup.
- **on_acquired** (Callable[[int], None] | None) - Optional - Callback invoked when the lock is acquired.

### Returns
None

### Notes
- This is a public constructor with one parameter per documented lock option.
- The `mode` parameter has no setter; construction is the only place it is set.
- When `thread_local=True`, configuration values (poll_interval, timeout, blocking, mode, lifetime) are per-thread and changes from one thread do not affect others.
- `lifetime` does not provide strict mutual exclusion for SoftFileLock.
- `close_error_policy` does not affect unlock failures or lock-file deletion.
- `fallback_to_soft` has no effect on Windows or SoftFileLock.
- `preserve_lock_file` affects Windows cleanup behavior (the text is cut off).
```

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

### Start internal method asynchronously

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/asyncio.html

Starts an internal method, handling both coroutine and synchronous callables. If the method is a coroutine function, it creates a task; otherwise, it runs it in the executor if configured, or directly sets the result on a new future.

```python
    def _start_internal_method(
        self, method: Callable[[], None] | Callable[[], Coroutine[None, None, None]]
    ) -> asyncio.Future[_BackendOutcome[None]]:
        if iscoroutinefunction(method):
            return asyncio.create_task(_capture_awaitable(cast("Callable[[], Coroutine[None, None, None]]", method)()))
        loop = asyncio.get_running_loop()
        sync_method = cast("Callable[[], None]", method)
        if self.run_in_executor:
            return loop.run_in_executor(self.executor, _capture_call, sync_method)
        future: asyncio.Future[_BackendOutcome[None]] = loop.create_future()
        future.set_result(_capture_call(sync_method))
        return future
```

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

### Read owner record from SoftFileLease

Source: https://py-filelock.readthedocs.io/en/latest/tutorials.html

SoftFileLease writes a richer protocol 2 record. Inside the with block, lease.owner returns an OwnerRecord with pid, hostname, mode, token, lease_duration, and start. The start field protects against PID reuse.

```python
from filelock import SoftFileLease

lease = SoftFileLease("server.lock", lease_duration=30)

with lease:
    if (owner := lease.owner) is not None:  # None if the marker is missing or unreadable
        print(owner.pid, owner.hostname)
        print(owner.mode)            # "lease"
        print(owner.token)           # names this particular claim
        print(owner.lease_duration)  # 30.0
        print(owner.start)           # process start token, or None on platforms without one
```

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

### Create SoftFileLease with lease duration

Source: https://py-filelock.readthedocs.io/en/latest/_sources/how-to.rst.txt

Create a SoftFileLease with a lease_duration of 30 seconds. The owner property returns an OwnerRecord because SoftFileLease reads the protocol 2 marker through MarkerSoftFileLock.

```python
from filelock import SoftFileLease

lease = SoftFileLease("work.lock", lease_duration=30)
```

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

### blocking

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_api.html

Property that gets or sets whether the locking is blocking or not.

```APIDOC
## Property: blocking

### Description
Gets or sets whether the locking is blocking or not.

### Getter Returns
- **bool** - True if blocking, False otherwise.

### Setter Parameters
- **value** (bool) - The new blocking value.

```

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

### poll_interval

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_api.html

Property that gets or sets the default polling interval in seconds.

```APIDOC
## Property: poll_interval

### Description
Gets or sets the default polling interval in seconds.

### Getter Returns
- **float** - The default polling interval in seconds.

### Setter Parameters
- **value** (float) - The new polling interval in seconds.

### Raises
- **ValueError** - If *value* is negative or not finite.
- **TypeError** - If *value* is not a real number.

```

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

### Using FileLock for cross-platform locking

Source: https://py-filelock.readthedocs.io/en/latest/_sources/concepts.rst.txt

Creates a FileLock instance, which automatically selects the best backend for the current platform (OS-level locking on Windows/Unix, soft-lock fallback without fcntl). Use this as the default choice for most applications.

```python
from filelock import FileLock

lock = FileLock("work.lock")
```

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

### timeout

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_api.html

Property that gets or sets the default timeout value in seconds.

```APIDOC
## Property: timeout

### Description
Gets or sets the default timeout value in seconds.

### Getter Returns
- **float** - The default timeout value in seconds.

### Setter Parameters
- **value** (float | str) - The new timeout value in seconds.

```

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

### Connect to SQLite with fork-safe settings

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_read_write.html

Creates a SQLite connection with fork-safe settings: check_same_thread=False, cached_statements=0, and a custom factory. Notes the SQLite use for fork tracking.

```python
def _connect(database: str, *, factory: type[_ForkSafeConnection], timeout: float) -> _ForkSafeConnection:
    _FORKED_DATABASES.note_sqlite_use()
    return sqlite3.connect(
        database,
        check_same_thread=False,
        factory=factory,
        cached_statements=0,
        timeout=timeout,
    )
```

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

### executor

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/asyncio.html

Property to get or set the executor used for running blocking operations.

```APIDOC
## executor

### Description
Get or set the executor used for running blocking operations.

### Method
@property
executor(self) -> futures.Executor | None

@executor.setter
def executor(self, value: futures.Executor | None) -> None

### Parameters
- **value** (futures.Executor | None) - Required - The new executor or `None`.
```

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

### AsyncSoftReadWriteLock.__init__

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_soft_rw/_async.html

Constructor for AsyncSoftReadWriteLock. Initializes the async wrapper with the provided lock file path and configuration options.

```APIDOC
## AsyncSoftReadWriteLock.__init__

### Description
Initializes an AsyncSoftReadWriteLock instance, wrapping a SoftReadWriteLock with async support.

### Method
Constructor

### Signature
```python
AsyncSoftReadWriteLock(
    lock_file: str | os.PathLike[str],
    timeout: float = -1,
    *,
    blocking: bool = True,
    is_singleton: bool = True,
    heartbeat_interval: float = 30.0,
    stale_threshold: float | None = None,
    poll_interval: float = 0.25,
    on_compromise: Callable[[LeaseCompromise], None] | None = None,
    loop: asyncio.AbstractEventLoop | None = None,
    executor: futures.Executor | None = None
) -> None
```

### Parameters
- **lock_file** (str | os.PathLike[str]) - Required - Path to the lock file; the protocol directory lives next to it as `<lock_file>.rw`
- **timeout** (float) - Optional - Maximum wait time in seconds; `-1` means block indefinitely. Default: `-1`
- **blocking** (bool) - Optional - If `False`, raise `Timeout` immediately on contention. Default: `True`
- **is_singleton** (bool) - Optional - If `True`, reuse existing `SoftReadWriteLock` instances per resolved path. Default: `True`
- **heartbeat_interval** (float) - Optional - Seconds between heartbeat refreshes. Default: `30.0`
- **stale_threshold** (float | None) - Optional - Seconds a holder record may stay unchanged before a contender evicts it. Defaults to `3 * heartbeat_interval`
- **poll_interval** (float) - Optional - Seconds between acquire retries under contention. Default: `0.25`
- **on_compromise** (Callable[[LeaseCompromise], None] | None) - Optional - Called from the heartbeat thread with a `LeaseCompromise` when the hold is lost
- **loop** (asyncio.AbstractEventLoop | None) - Optional - Event loop for `run_in_executor`; `None` uses the running loop
- **executor** (futures.Executor | None) - Optional - Executor for `run_in_executor`; `None` uses the default executor

### Returns
None
```

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

### Use AsyncReadWriteLock

Source: https://py-filelock.readthedocs.io/en/latest/_sources/how-to.rst.txt

Example of using AsyncReadWriteLock for asynchronous read and write operations.

```python
from filelock import AsyncReadWriteLock

rw = AsyncReadWriteLock("data.db")

async with rw.read_lock():
    data = await get_shared_data()

async with rw.write_lock():
    await update_shared_data()
```

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

### Verify singleton lookup with get_lock

Source: https://py-filelock.readthedocs.io/en/latest/_sources/how-to.rst.txt

Demonstrates that get_lock is a sugar for the same singleton lookup as the constructor.

```python
assert SoftReadWriteLock.get_lock("/shared/nfs/data.lock") is SoftReadWriteLock("/shared/nfs/data.lock")
```

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

### acquire_read

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_soft_rw/_sync.html

Acquires a shared read lock, supporting reentrant acquisition and starting a heartbeat thread on first acquisition.

```APIDOC
## acquire_read

### Description
Acquire a shared read lock. If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a read lock while holding a write lock raises RuntimeError. On the 0→1 transition, the reader publishes its holder record, enters the next snapshot once no live writer is named, and starts a daemon heartbeat thread that rewrites the record's nonce every heartbeat_interval seconds.

### Method
Method (Python)

### Endpoint
acquire_read(timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy

### Parameters
- **timeout** (float | None) - Optional - Maximum wait time in seconds, or None to use the instance default; -1 means block indefinitely.
- **blocking** (bool | None) - Optional - If False, raise Timeout immediately when the lock is unavailable; None uses the instance default.

### Usage
proxy = lock.acquire_read(timeout=10, blocking=True)
# Use the resource
try:
    # ...
finally:
    lock.release()

```

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

### lifetime

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_api.html

Property that gets or sets the soft marker age in seconds that permits expiry, or None to disable age-based expiry.

```APIDOC
## Property: lifetime

### Description
Gets or sets the soft marker age in seconds that permits expiry, or None to disable age-based expiry.

### Getter Returns
- **float | None** - The soft marker age in seconds, or None if disabled.

### Setter Parameters
- **value** (float | None) - The new lifetime value in seconds, or None to disable expiration.

### Raises
- **ValueError** - If *value* is negative or not finite.
- **TypeError** - If *value* is not None and not a real number.

```

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

### Manage acquisition defaults

Source: https://py-filelock.readthedocs.io/en/latest/_sources/how-to.rst.txt

Set instance-wide defaults for timeout and blocking, and override them per call.

```python
rw = SoftReadWriteLock("/shared/nfs/data.lock", timeout=30, blocking=True)

with rw.read_lock():  # inherits timeout=30
    pass

with rw.read_lock(timeout=-1):  # blocks forever, overriding the instance default
    pass

with rw.read_lock(blocking=False):  # one attempt; ignores timeout entirely
    pass
```

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

### TOCTOU vulnerability example

Source: https://py-filelock.readthedocs.io/en/latest/_sources/concepts.rst.txt

Shows a TOCTOU vulnerability where a file is checked for existence and then opened, leaving a race window for symlink attacks.

```python
if os.path.exists("sensitive.txt"):  # Check
    data = open("sensitive.txt").read()  # Use (race window here!)
```

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

### Manual acquire and release with try/finally

Source: https://py-filelock.readthedocs.io/en/latest/_sources/tutorials.rst.txt

Use lock.acquire() and lock.release() with a try/finally block to guarantee the lock is released even if an exception occurs.

```python
lock.acquire()
try:
    print("I have the lock")
finally:
    lock.release()
```

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

### OwnerRecord

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_marker.html

A named tuple representing the owner published in a protocol 2 marker. Fields include pid, hostname, mode, token, lease_duration, and start.

```APIDOC
## OwnerRecord

### Description
A NamedTuple that represents the owner record published in a protocol 2 marker. It contains identifying information about the process that holds the lock.

### Fields
- **pid** (int) - Required - The process ID of the lock owner.
- **hostname** (str) - Required - The hostname of the machine where the lock is held.
- **mode** (OwnerMode) - Required - The owner mode, either 'lease' or 'unknown'.
- **token** (str | None) - Optional - An identifier token for the lock.
- **lease_duration** (float | None) - Optional - The duration of the lease in seconds.
- **start** (int | None) - Optional - The start token of the process.

### Example
```python
OwnerRecord(pid=1234, hostname='myhost', mode='lease')
```
```

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

### Usage of acquire() in context manager and try-finally

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/asyncio.html

Shows two equivalent ways to use the acquire() method: with a context manager (recommended) and with an explicit try-finally construct. The context manager automatically releases the lock, while the try-finally requires manual release.

```python
# You can use this method in the context manager (recommended)
with lock.acquire():
    pass

# Or use an equivalent try-finally construct:
lock.acquire()
try:
    pass
finally:
    lock.release()
```

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

### Lock selection flowchart (Mermaid)

Source: https://py-filelock.readthedocs.io/en/latest/_sources/concepts.rst.txt

A mermaid diagram that guides the user through choosing a lock type based on workload, filesystem, and concurrency requirements. It includes custom CSS class definitions for styling the diagram nodes.

```mermaid
flowchart TD
    start["Choose a lock type"] --> question1{"Read-heavy workload?"}
    question1 -->|Yes| questionRwNet{"Network<br/>filesystem?"}
    questionRwNet -->|Yes| questionRwFs{"Verified marker and<br/>cache behavior?"}
    questionRwFs -->|Yes| srw["Use SoftReadWriteLock"]
    questionRwFs -->|No| service["Use a lock service"]
    questionRwNet -->|No| questionAsync{"Async code?"}
    questionAsync -->|Yes| arw["Use AsyncReadWriteLock"]
    questionAsync -->|No| rw["Use ReadWriteLock"]
    question1 -->|No| question2{"Need network<br/>filesystem support?"}
    question2 -->|Yes| questionSoftFs{"Verified exclusive create<br/>and cache behavior?"}
    questionSoftFs -->|No| service
    questionSoftFs -->|Yes| questionContract{"Which soft-lock<br/>contract?"}
    questionContract -->|"Strict, never overlap"| strict["Use StrictSoftFileLock"]
    questionContract -->|"Overlap OK, fence resource"| lease["Use SoftFileLease"]
    questionContract -->|"Cooperative best-effort"| soft["Use SoftFileLock"]
    question2 -->|No| question3{"Need platform<br/>specific control?"}
    question3 -->|Yes| platform["Use UnixFileLock<br/>or WindowsFileLock"]
    question3 -->|No| default["Use FileLock<br/>(recommended)"]

    %% docs/_static/custom.css colors these classes, once per light and dark theme
    classDef recommended stroke-width:2px
    classDef alternative stroke-width:2px
    classDef special stroke-width:2px
    classDef escape stroke-width:2px
    class default recommended
    class soft,strict,lease,platform,srw alternative
    class rw,arw special
    class service escape
```

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

### Low-level acquire and release methods for ReadWriteLock

Source: https://py-filelock.readthedocs.io/en/latest/_sources/how-to.rst.txt

Shows using acquire_read() and acquire_write() with try/finally to ensure release(). Provides more control than context managers.

```python
rw.acquire_read(timeout=5)
try:
    data = get_shared_data()
finally:
    rw.release()

rw.acquire_write(timeout=5)
try:
    update_shared_data()
finally:
    rw.release()
```

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

### Create a basic FileLock

Source: https://py-filelock.readthedocs.io/en/latest/_sources/tutorials.rst.txt

Creates a FileLock object representing a lock file on disk. The lock is not acquired until used with a context manager or acquire().

```python
from filelock import FileLock

lock = FileLock("cache/.locks/models/bert-base/a1b2c3.lock")
```

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

### Correct lock file usage

Source: https://py-filelock.readthedocs.io/en/latest/_sources/concepts.rst.txt

Shows the correct approach: using a separate lock file to protect the target file, ensuring the lock remains valid even if the target is deleted.

```python
lock = FileLock("data.txt.lock")  # ✓ Right
with lock:
    data = open("data.txt").read()
```

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

### filelock._soft_rw._sync: _do_acquire_inner method

Source: https://py-filelock.readthedocs.io/en/latest/_modules/filelock/_soft_rw/_sync.html

Acquires the lock with reentrancy validation, publishes a participant, waits for the participant to advance, and starts a heartbeat thread. On failure, it leaves the participant and re-raises. Raises RuntimeError for invalid reentrant mode changes or cross-thread write acquisition.

```python
    def _do_acquire_inner(
        self,
        mode: Mode,
        effective_timeout: float,
        start: float,
        *,
        blocking: bool,
    ) -> AcquireReturnProxy:
        with self._locks.internal:
            if self._hold is not None:
                return self._validate_reentrant(mode)
        deadline = None if effective_timeout == -1 else start + effective_timeout
        participant = Participant(
            self._files,
            self._root,
            mode,
            stale_threshold=self.stale_threshold,
            clock=time.monotonic,
            ledger=self._ledger,
            log=self._log,
        )
        participant.publish()
        try:
            self._wait_for(participant.advance, deadline=deadline, blocking=blocking)
        except BaseException:
            # A contender that gives up must not stay named in the snapshot: a writer left there would block every
            # reader until a peer waits out the stale threshold. Its own token is all leave() touches, so it cannot
            # undo a peer's claim.
            with suppress(OSError):
                participant.leave()
            raise
        stop_event = threading.Event()
        hold = _Hold(
            level=1,
            mode=mode,
            write_thread_id=threading.get_ident() if mode == "write" else None,
            participant=participant,
            heartbeat_thread=_HeartbeatThread(
                refresh=lambda: self._refresh(participant, hold),
                interval=self.heartbeat_interval,
                stop_event=stop_event,
                name=f"filelock-heartbeat-{id(self):x}",
            ),
            heartbeat_stop=stop_event,
            last_refresh=time.monotonic(),
        )
        # Publish the hold and start its heartbeat under one internal-lock section, so a concurrent release() never
        # observes a hold whose thread has not started and joins it. If the OS refuses the thread, clear the hold and
        # leave: left in place, a peer evicts the unrefreshed record and acquires while this instance still believes it
        # holds the lock.
        start_error: BaseException | None = None
        with self._locks.internal:
            self._hold = hold
            self._compromise = None
            try:
                hold.heartbeat_thread.start()
            except BaseException as error:  # ruff:ignore[blind-except]  # clear the slot below and re-raise
                self._hold = None
                start_error = error
        if start_error is not None:
            participant.leave()
            raise start_error
        return AcquireReturnProxy(lock=self)
```
Tokens:
Raw