52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""同一本地数据库的 Windows named mutex。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from cmbuyer_client.core.errors import SingleInstanceError
|
|
|
|
|
|
class NamedMutex:
|
|
_ALREADY_EXISTS = 183
|
|
|
|
def __init__(self, database_path: Path) -> None:
|
|
if os.name != "nt":
|
|
raise SingleInstanceError("named_mutex_requires_windows")
|
|
canonical = str(database_path.expanduser().resolve()).casefold().encode("utf-8")
|
|
# Global namespace 覆盖同一 Windows 用户的多个交互 session；默认 DACL 不向其他用户泄露句柄。
|
|
name = "Global\\cmbuyer-" + hashlib.sha256(canonical).hexdigest()
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
kernel32.CreateMutexW.argtypes = (wintypes.LPVOID, wintypes.BOOL, wintypes.LPCWSTR)
|
|
kernel32.CreateMutexW.restype = wintypes.HANDLE
|
|
kernel32.ReleaseMutex.argtypes = (wintypes.HANDLE,)
|
|
kernel32.ReleaseMutex.restype = wintypes.BOOL
|
|
kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
|
|
kernel32.CloseHandle.restype = wintypes.BOOL
|
|
ctypes.set_last_error(0)
|
|
handle = kernel32.CreateMutexW(None, True, name)
|
|
if not handle:
|
|
raise SingleInstanceError("named_mutex_failed")
|
|
if ctypes.get_last_error() == self._ALREADY_EXISTS:
|
|
kernel32.CloseHandle(handle)
|
|
raise SingleInstanceError("instance_already_running")
|
|
self._kernel32 = kernel32
|
|
self._handle = handle
|
|
|
|
def close(self) -> None:
|
|
handle = getattr(self, "_handle", None)
|
|
if handle:
|
|
self._kernel32.ReleaseMutex(handle)
|
|
self._kernel32.CloseHandle(handle)
|
|
self._handle = None
|
|
|
|
def __enter__(self) -> "NamedMutex":
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
|
|
self.close()
|