feat: optimize decorators

eth_portfolio/_db/decorators.py

@@ -1,62 +1,80 @@
-import asyncio
-import logging
-import time
+from asyncio import iscoroutinefunction
+from asyncio import sleep as aio_sleep
 from functools import wraps
+from logging import DEBUG, getLogger
 from random import random
+from time import sleep as time_sleep
 from typing import Callable, TypeVar
 
 from a_sync._typing import AnyFn
 from pony.orm import OperationalError, TransactionError
 from typing_extensions import ParamSpec
 
+
 P = ParamSpec("P")
 T = TypeVar("T")
 
-logger = logging.getLogger(__name__)
+
+logger = getLogger(__name__)
+__logger_is_enabled_for = logger.isEnabledFor
+__logger_warning = logger.warning
+__logger_log = logger._log
 
 
 def break_locks(fn: AnyFn[P, T]) -> AnyFn[P, T]:
-    if asyncio.iscoroutinefunction(fn):
+    if iscoroutinefunction(fn):
 
         @wraps(fn)
         async def break_locks_wrap(*args: P.args, **kwargs: P.kwargs) -> T:
+            debug_logs_enabled = None
             tries = 0
             while True:
                 try:
                     return await fn(*args, **kwargs)
                 except OperationalError as e:
-                    logger.debug("%s.%s %s", fn.__module__, fn.__name__, e)
                     if str(e) != "database is locked":
                         raise e
-                    sleep = tries * random()
-                    await asyncio.sleep(sleep)
+
+                    if debug_logs_enabled is None:
+                        debug_logs_enabled = __logger_is_enabled_for(DEBUG)
+
+                    if debug_logs_enabled is True:
+                        __logger_log(DEBUG, "%s.%s %s", (fn.__module__, fn.__name__, e))
+
+                    await aio_sleep(tries * random())
                     tries += 1
                     if tries > 5:
-                        logger.warning("%s caught in err loop with %s", fn, e)
+                        __logger_warning("%s caught in err loop with %s", fn, e)
 
     else:
 
         @wraps(fn)
         def break_locks_wrap(*args: P.args, **kwargs: P.kwargs) -> T:
+            debug_logs_enabled = None
             tries = 0
             while True:
                 try:
                     return fn(*args, **kwargs)  # type: ignore [return-value]
                 except OperationalError as e:
-                    logger.debug("%s.%s %s", fn.__module__, fn.__name__, e)
                     if str(e) != "database is locked":
                         raise e
-                    sleep = tries * random()
-                    time.sleep(sleep)
+
+                    if debug_logs_enabled is None:
+                        debug_logs_enabled = __logger_is_enabled_for(DEBUG)
+
+                    if debug_logs_enabled is True:
+                        __logger_log(DEBUG, "%s.%s %s", (fn.__module__, fn.__name__, e))
+
+                    time_sleep(tries * random())
                     tries += 1
                     if tries > 5:
-                        logger.warning("%s caught in err loop with %s", fn, e)
+                        __logger_warning("%s caught in err loop with %s", fn, e)
 
     return break_locks_wrap
 
 
 def requery_objs_on_diff_tx_err(fn: Callable[P, T]) -> Callable[P, T]:
-    if asyncio.iscoroutinefunction(fn):
+    if iscoroutinefunction(fn):
         raise TypeError(f"{fn} must not be async")
 
     @wraps(fn)