mirror of
https://github.com/bottlesdevs/Bottles.git
synced 2026-10-28 23:14:14 +08:00
79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
# decorators.py
|
|
#
|
|
# Copyright 2022 brombinmirko <send@mirko.pm>
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, in version 3 of the License.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
|
|
from functools import lru_cache, wraps
|
|
from threading import Lock as PyLock
|
|
from time import monotonic_ns
|
|
from typing import Callable, Dict
|
|
|
|
|
|
def cache(_func=None, *, seconds: int = 600, maxsize: int = 128, typed: bool = False):
|
|
"""
|
|
Extension of functools lru_cache with a timeout
|
|
|
|
Parameters:
|
|
seconds (int): Timeout in seconds to clear the WHOLE cache, default = 10 minutes
|
|
maxsize (int): Maximum Size of the Cache
|
|
typed (bool): Same value of different type will be a different entry
|
|
|
|
Source: <https://gist.github.com/Morreski/c1d08a3afa4040815eafd3891e16b945>
|
|
"""
|
|
|
|
def wrapper_cache(f):
|
|
f = lru_cache(maxsize=maxsize, typed=typed)(f)
|
|
f.delta = seconds * 10 ** 9
|
|
f.expiration = monotonic_ns() + f.delta
|
|
|
|
@wraps(f)
|
|
def wrapped_f(*args, **kwargs):
|
|
if monotonic_ns() >= f.expiration:
|
|
f.cache_clear()
|
|
f.expiration = monotonic_ns() + f.delta
|
|
return f(*args, **kwargs)
|
|
|
|
wrapped_f.cache_info = f.cache_info
|
|
wrapped_f.cache_clear = f.cache_clear
|
|
return wrapped_f
|
|
|
|
# To allow decorator to be used without arguments
|
|
if _func is None:
|
|
return wrapper_cache
|
|
else:
|
|
return wrapper_cache(_func)
|
|
|
|
|
|
class Lock:
|
|
LOCKS: Dict[str, PyLock] = {}
|
|
|
|
@staticmethod
|
|
def get_mutex_lock(name: str) -> PyLock:
|
|
Lock.LOCKS.setdefault(name, PyLock())
|
|
return Lock.LOCKS[name]
|
|
|
|
@staticmethod
|
|
def mutex(name: str):
|
|
Lock.LOCKS.setdefault(name, PyLock())
|
|
|
|
def func_wrapper(func: Callable):
|
|
def wrapper(*args, **kwargs):
|
|
Lock.LOCKS[name].acquire()
|
|
rv = func(*args, **kwargs)
|
|
Lock.LOCKS[name].release()
|
|
return rv
|
|
return wrapper
|
|
return func_wrapper
|