✨ feat: add useSynchronized

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "usepy"
-version = "0.2.8"
+version = "0.2.9"
 description = "usepy"
 homepage = "https://usepy.code05.com/"
 authors = ["miclon <jcnd@163.com>"]

src/usepy/__init__.py

@@ -9,6 +9,7 @@
     "useAdDict",
     "useBloomFilter",
     "useCachedProperty",
+    "useSynchronized",
     "useCatchError",
     "useCleanHtml",
     "useCounter",

src/usepy/core/useSynchronized.py

@@ -0,0 +1,21 @@
+import functools
+from threading import Lock
+
+
+def useSynchronized(func):
+    """
+    线程安全装饰器
+    >>> @useSynchronized
+    >>> def demo(x):
+    >>>     return x
+    """
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        lock = vars(func).get("_synchronized_lock", None)
+        if lock is None:
+            lock = vars(func).setdefault("_synchronized_lock", Lock())
+        with lock:
+            return func(*args, **kwargs)
+
+    return wrapper
+

tests/test_core/test_useSynchronized.py

@@ -0,0 +1,30 @@
+import pytest
+from usepy import useSynchronized
+import threading
+
+
+@useSynchronized
+def add(x, y):
+    return x + y
+
+
+def test_useSynchronized_decorator():
+
+    def thread_function(result_list):
+        result = add(1, 2)
+        result_list.append(result)
+
+    results = []
+    threads = []
+    num_threads = 10
+
+    for _ in range(num_threads):
+        thread = threading.Thread(target=thread_function, args=(results,))
+        threads.append(thread)
+        thread.start()
+
+    for thread in threads:
+        thread.join()
+
+    assert all(result == 3 for result in results)
+