portalocker.utils module

class portalocker.utils.Lock(filename: Union[str, pathlib.Path], mode: str = 'a', timeout: float = 5, check_interval: float = 0.25, fail_when_locked: bool = False, flags: portalocker.constants.LockFlags = LockFlags.None, **file_open_kwargs)[source]

Bases: portalocker.utils.LockBase

Lock manager with build-in timeout

  • filename – filename

  • mode – the open mode, ‘a’ or ‘ab’ should be used for writing

  • truncate – use truncate to emulate ‘w’ mode, None is disabled, 0 is truncate to 0 bytes

  • timeout – timeout when trying to acquire a lock

  • check_interval – check interval while waiting

  • fail_when_locked – after the initial lock failed, return an error or lock the file. This does not wait for the timeout.

  • **file_open_kwargs – The kwargs for the open(…) call

fail_when_locked is useful when multiple threads/processes can race when creating a file. If set to true than the system will wait till the lock was acquired and then return an AlreadyLocked exception.

Note that the file is opened first and locked later. So using ‘w’ as mode will result in truncate _BEFORE_ the lock is checked.

acquire(timeout: Optional[float] = None, check_interval: Optional[float] = None, fail_when_locked: Optional[bool] = None) IO[source]

Acquire the locked filehandle

check_interval: float

check interval while waiting for timeout

fail_when_locked: bool

skip the timeout and immediately fail if the initial lock fails

fh: Optional[IO]
filename: str
flags: portalocker.constants.LockFlags
mode: str

Releases the currently locked file handle

timeout: float

timeout when trying to acquire a lock

truncate: bool
portalocker.utils.open_atomic(filename: Union[str, pathlib.Path], binary: bool = True)[source]

Open a file for atomic writing. Instead of locking this method allows you to write the entire file and move it to the actual location. Note that this makes the assumption that a rename is atomic on your platform which is generally the case but not a guarantee.


>>> filename = 'test_file.txt'
>>> if os.path.exists(filename):
...     os.remove(filename)
>>> with open_atomic(filename) as fh:
...     written = fh.write(b'test')
>>> assert os.path.exists(filename)
>>> os.remove(filename)
>>> import pathlib
>>> path_filename = pathlib.Path('test_file.txt')
>>> with open_atomic(path_filename) as fh:
...     written = fh.write(b'test')
>>> assert path_filename.exists()
>>> path_filename.unlink()