safe_contrainer.py 814 Bytes
import threading
from typing import TypeVar, List, Dict, Generic

K = TypeVar('K')  # 定义类型变量

V = TypeVar('V')


class SafeDict(Generic[K, V]):

    def __init__(self):
        self.lock = threading.Lock()
        self.data: Dict[K, V] = {}

    def put(self, key: K, value: V):
        with self.lock:
            self.data.update({key: value})

    def remove(self, key: K):
        with self.lock:
            self.data.pop(key)

    def values(self) -> List[V]:
        with self.lock:
            return list(self.data.values())

    def get(self, key: K) -> V:
        with self.lock:
            return self.data.get(key)




if __name__ == '__main__':
    test: SafeDict[int, str] = SafeDict()
    test.put(1, '1')
    print(test.get(1))
    print(test.get(2))
    t2 = {}
    print(t2.get(1))