pylocked provides utility classes (and functions) to lock any resource (or function) and thus make them thread and / or concurrency safe.
From pip:
pip install pylockedFrom repository:
pip install git+https://github.com/Eyal-Shalev/pylockedfrom pylocked.asyncio import AsyncLocked
locked_arr: Locked[list[int]] = Locked([])
async def double():
async with locked_arr as arr:
for i in range(len(arr)):
arr[i] += arr[i]
async def reset():
await locked_arr.replace(list(range(10)))
async def duplicate():
await locked_arr.update(lambda arr: arr * 2)from typing import Optional
from pylocked.asyncio import locked
class LockedSingleton:
_instance: Optional[LockedSingleton] = None
@locked
@staticmethod
async def get_instance() -> LockedSingleton:
if LockedSingleton._instance is None:
LockedSingleton._instance = LockedSingleton()
return LockedSingleton._instanceYou can use
pylocked.threading.Lockedif you want to usethreading.Lockinstead.
from pylocked.threading import RLocked
locked_arr: RLocked[list[int]] = RLocked([])
def double():
with locked_arr as arr:
for i in range(len(arr)):
arr[i] += arr[i]
def reset():
locked_arr.replace(list(range(10)))
def duplicate():
locked_arr.update(lambda arr: arr * 2)You can use
pylocked.threading.locked()if you want to usethreading.Lockinstead.
from typing import Optional
from pylocked.threading import rlocked
class LockedSingleton:
_instance: Optional[LockedSingleton] = None
@rlocked
@staticmethod
def get_instance() -> LockedSingleton:
if LockedSingleton._instance is None:
LockedSingleton._instance = LockedSingleton()
return LockedSingleton._instance
