GeezerVison is showing github.com — pre-commit/pre_commit/file_lock.py at main · arshdeepkumar482/pre-commit

Expand file tree

file_lock.py

Copy path

More file actions

Latest commit

History

75 lines (65 loc) · 2.31 KB

file_lock.py

Copy path

File metadata and controls

75 lines (65 loc) · 2.31 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

from __future__ import annotations

import contextlib

import errno

import sys

from typing import Callable

from typing import Generator

if sys.platform == 'win32': # pragma: no cover (windows)

import msvcrt

# https://docs.microsoft.com/en-us/cpp/c-runtime-library/reference/locking

# on windows we lock "regions" of files, we don't care about the actual

# byte region so we'll just pick *some* number here.

_region = 0xffff

@contextlib.contextmanager

def _locked(

fileno: int,

blocked_cb: Callable[[], None],

) -> Generator[None, None, None]:

try:

msvcrt.locking(fileno, msvcrt.LK_NBLCK, _region)

except OSError:

blocked_cb()

while True:

try:

msvcrt.locking(fileno, msvcrt.LK_LOCK, _region)

except OSError as e:

# Locking violation. Returned when the _LK_LOCK or _LK_RLCK

# flag is specified and the file cannot be locked after 10

# attempts.

if e.errno != errno.EDEADLOCK:

raise

else:

break

try:

yield

finally:

# From cursory testing, it seems to get unlocked when the file is

# closed so this may not be necessary.

# The documentation however states:

# "Regions should be locked only briefly and should be unlocked

# before closing a file or exiting the program."

msvcrt.locking(fileno, msvcrt.LK_UNLCK, _region)

else: # pragma: win32 no cover

import fcntl

@contextlib.contextmanager

def _locked(

fileno: int,

blocked_cb: Callable[[], None],

) -> Generator[None, None, None]:

try:

fcntl.flock(fileno, fcntl.LOCK_EX | fcntl.LOCK_NB)

except OSError: # pragma: no cover (tests are single-threaded)

blocked_cb()

fcntl.flock(fileno, fcntl.LOCK_EX)

try:

yield

finally:

fcntl.flock(fileno, fcntl.LOCK_UN)

@contextlib.contextmanager

def lock(

path: str,

blocked_cb: Callable[[], None],

) -> Generator[None, None, None]:

with open(path, 'a+') as f:

with _locked(f.fileno(), blocked_cb):

yield