Fix file locking on Windows using msvcrt instead of fcntl

src/rigbook/db.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import sys
 import uuid as _uuid
 from datetime import datetime, timezone
 from pathlib import Path
@@ -15,6 +16,34 @@
 DB_DIR = Path.home() / ".local" / "rigbook"
 
 
+def _lock_exclusive(f) -> None:
+    """Acquire a non-blocking exclusive lock on a file (cross-platform)."""
+    if sys.platform == "win32":
+        import msvcrt
+
+        msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
+    else:
+        import fcntl
+
+        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+
+def _unlock(f) -> None:
+    """Release a file lock (cross-platform)."""
+    if sys.platform == "win32":
+        import msvcrt
+
+        try:
+            f.seek(0)
+            msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
+        except OSError:
+            pass
+    else:
+        import fcntl
+
+        fcntl.flock(f, fcntl.LOCK_UN)
+
+
 class Base(DeclarativeBase):
     pass
 
@@ -178,15 +207,13 @@ def default_db_path(self) -> Path:
 
     def check_lock(self, db_path: Path) -> None:
         """Raise DatabaseLockError if the database is locked by another process."""
-        import fcntl
-
         lock_path = db_path.with_suffix(".lock")
         if not lock_path.exists():
             return
         try:
             with open(lock_path, "r+") as f:
-                fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
-                fcntl.flock(f, fcntl.LOCK_UN)
+                _lock_exclusive(f)
+                _unlock(f)
         except OSError:
             raise DatabaseLockError(
                 f"Logbook '{db_path.stem}' is already open in another process"
@@ -233,12 +260,10 @@ def _write_lock_content(self) -> None:
 
     def _acquire_lock(self, db_path: Path) -> None:
         """Acquire an exclusive file lock to prevent concurrent access."""
-        import fcntl
-
         lock_path = db_path.with_suffix(".lock")
         self._lock_file = open(lock_path, "w")
         try:
-            fcntl.flock(self._lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            _lock_exclusive(self._lock_file)
             self._write_lock_content()
         except OSError:
             self._lock_file.close()
@@ -249,11 +274,9 @@ def _acquire_lock(self, db_path: Path) -> None:
 
     def _release_lock(self) -> None:
         """Release the file lock."""
-        import fcntl
-
         if self._lock_file:
             try:
-                fcntl.flock(self._lock_file, fcntl.LOCK_UN)
+                _unlock(self._lock_file)
                 lock_path = Path(self._lock_file.name)
                 self._lock_file.close()
                 lock_path.unlink(missing_ok=True)