From 267ac5c3caf1052dcaca0f299a1d38b10493d020 Mon Sep 17 00:00:00 2001
From: "google-labs-jules[bot]"
 <161369871+google-labs-jules[bot]@users.noreply.github.com>
Date: Mon, 16 Feb 2026 17:43:29 +0000
Subject: [PATCH] Refactor src/utils/file_lock.py to reduce duplication

Extracts common logic for path preparation and lock acquisition into a private helper function `_get_portalocker_lock`. This simplifies `exclusive_lock` and `shared_lock` implementations.

Also updates the docstring for `exclusive_lock` to correctly describe the file mode as "binary read/write mode" (r+b) instead of "binary append mode".

Co-authored-by: PR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com>
---
 src/utils/file_lock.py | 52 ++++++++++++++++--------------------------
 1 file changed, 20 insertions(+), 32 deletions(-)

diff --git a/src/utils/file_lock.py b/src/utils/file_lock.py
index f2c1aa13..0cbc0203 100644
--- a/src/utils/file_lock.py
+++ b/src/utils/file_lock.py
@@ -7,34 +7,36 @@ from pathlib import Path
 import portalocker
 
 
+def _get_portalocker_lock(
+    path: Path, flags: int, timeout: Optional[float] = None
+) -> portalocker.Lock:
+    """Helper to prepare path and create a portalocker.Lock instance."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.touch(exist_ok=True)
+
+    kwargs = {
+        "mode": "r+b",
+        "flags": flags,
+    }
+    if timeout is not None:
+        kwargs["timeout"] = timeout
+
+    return portalocker.Lock(str(path), **kwargs)
+
+
 @contextmanager
 def exclusive_lock(
     path: Union[str, PathLike[str], Path], timeout: Optional[float] = None
 ) -> Generator[None, None, None]:
     """Context manager that locks *path* exclusively.
 
-    The function opens the file in binary append mode and obtains an
+    The function opens the file in binary read/write mode and obtains an
     exclusive lock using ``portalocker``. If ``timeout`` is provided,
     acquiring the lock will wait for at most that many seconds before
     raising ``portalocker.exceptions.LockException``.
     """
     path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.touch(exist_ok=True)
-    flags = portalocker.LockFlags.EXCLUSIVE
-    if timeout is None:
-        lock = portalocker.Lock(
-            str(path),
-            mode="r+b",
-            flags=flags,
-        )
-    else:
-        lock = portalocker.Lock(
-            str(path),
-            mode="r+b",
-            timeout=timeout,
-            flags=flags,
-        )
+    lock = _get_portalocker_lock(path, portalocker.LockFlags.EXCLUSIVE, timeout)
     with lock as fh:
         yield fh
 
@@ -51,21 +53,7 @@ def shared_lock(
     ``portalocker.exceptions.LockException``.
     """
     path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.touch(exist_ok=True)
-    if timeout is None:
-        lock = portalocker.Lock(
-            str(path),
-            mode="r+b",
-            flags=portalocker.LockFlags.SHARED,
-        )
-    else:
-        lock = portalocker.Lock(
-            str(path),
-            mode="r+b",
-            timeout=timeout,
-            flags=portalocker.LockFlags.SHARED,
-        )
+    lock = _get_portalocker_lock(path, portalocker.LockFlags.SHARED, timeout)
     with lock as fh:
         fh.seek(0)
         yield fh