Async guard

This should prevent race condition and crashes when when changing modes (locking, unlocking, creating new profile)

clive/__private/core/async_guard.py

+from __future__ import annotations
+
+import asyncio
+import contextlib
+from typing import TYPE_CHECKING, Final, Generator
+
+from clive.exceptions import CliveError
+
+if TYPE_CHECKING:
+    from types import TracebackType
+
+
+class AsyncGuardNotAvailableError(CliveError):
+    """Raised when trying to acquire a guard that is already acquired."""
+
+    MESSAGE: Final[str] = "Guard is already acquired."
+
+    def __init__(self) -> None:
+        super().__init__(self.MESSAGE)
+
+
+class AsyncGuard:
+    """
+    A helper class to manage an asynchronous event-like lock, ensuring exclusive execution.
+
+    Use this for scenarios where you want to prevent concurrent execution of an async task.
+    When the guard is acquired by some other task, the guarded block could not execute, error will be raised instead.
+
+    Can be used together with `suppress`. Look into its documentation for more details.
+
+    Usage:
+        ```
+        async_guard = AsyncGuard()
+
+        with async_guard:
+            # Protected code which shouldn't be executed concurrently
+        ```
+    """
+
+    def __init__(self) -> None:
+        self._event = asyncio.Event()
+
+    def __enter__(self) -> None:
+        if not self.is_available:
+            raise AsyncGuardNotAvailableError
+
+        self.acquire()
+
+    def __exit__(self, _: type[BaseException] | None, ex: BaseException | None, ___: TracebackType | None) -> None:
+        self.release()
+
+    @property
+    def is_available(self) -> bool:
+        """
+        Check if the event lock is currently available (not acquired).
+
+        Use this to determine if an instruction can proceed with no conflicts.
+        """
+        return not self._event.is_set()
+
+    def acquire(self) -> None:
+        self._event.set()
+
+    def release(self) -> None:
+        self._event.clear()
+
+    @staticmethod
+    @contextlib.contextmanager
+    def suppress() -> Generator[None]:
+        """
+        Suppresses the AsyncGuardNotAvailable error raised by the guard.
+
+        Use this together with `guard` to skip code execution when like below.
+
+        Usage:
+            ```
+            async_guard = AsyncGuard()
+
+            with async_guard.suppress(), async_guard:
+                # Code that should be skipped when guard is acquired
+            ```
+        """
+        with contextlib.suppress(AsyncGuardNotAvailableError):
+            yield