Python equivalent of Java synchronized

You can write your own @synchronized decorator.

The example uses a Mutex Lock:

from functools import wraps
from multiprocessing import Lock


def synchronized(member):
    """
    @synchronized decorator.

    Lock a method for synchronized access only. The lock is stored to
    the function or class instance, depending on what is available.
    """

    @wraps(member)
    def wrapper(*args, **kwargs):
        lock = vars(member).get("_synchronized_lock", None)
        result = None
        try:
            if lock is None:
                lock = vars(member).setdefault("_synchronized_lock", Lock())
            lock.acquire()
            result = member(*args, **kwargs)
            lock.release()
        except Exception as e:
            lock.release()
            raise e
        return result

    return wrapper

Now your are able to decorate a method like this:

class MyClass:
  ...
  @synchronized
  def hello_world(self):
    print("synced hello world")

And there is also an excellent Blog post about the missing synchronized decorator.

Working code using with self.lock which can take care of exception if occurs:

Inside Manager we are making Manager mehods thread safe :

from threading import RLock


class Manager:
    def __init__(self):
        self.lock = RLock()
        self.hash: dict[str, int] = dict()

    def containsToken(self, key) -> bool:
        with self.lock:
            self.lock.acquire()
            return key in self.hash


    def addToken(self, token: str):
        with self.lock:
            token = token.strip()
            if token in self.hash:
                self.hash[token] = self.hash[token] + 1
            else:
                self.hash[token] = 1


    def removeToken(self, token):
        with self.lock:
            if token not in self.hash:
                raise KeyError(f"token : {token} doesn't exits")
            self.hash[token] = self.hash[token] - 1
            if self.hash[token] == 0:
                self.hash.pop(token)

if __name__ == "__main__":
    sync = Manager()
    sync.addToken("a")
    sync.addToken("a")
    sync.addToken("a")
    sync.addToken("a")
    sync.addToken("B")
    sync.addToken("B")
    sync.addToken("B")
    sync.addToken("B")
    sync.removeToken("a")
    sync.removeToken("a")
    sync.removeToken("a")
    sync.removeToken("B")
    print(sync.hash)

Output:

{'a': 1, 'B': 3}