add timed cache decorator

app/core/bot.py

@@ -4,7 +4,7 @@ from aiogram import Bot, types
 from aiogram.contrib.middlewares.logging import LoggingMiddleware
 from aiogram.dispatcher import Dispatcher
 from aiogram.utils.callback_data import CallbackData
-from app.core.parse_web import get_driver, get_ttl_hash, parse_site
+from app.core.parse_web import get_driver, parse_site
 from app.settings import TELEGRAM_API_TOKEN
 
 bot = Bot(token=TELEGRAM_API_TOKEN)
@@ -36,7 +36,7 @@ def get_keyboard() -> types.InlineKeyboardMarkup:
 async def home_office(
     query: types.CallbackQuery, callback_data: dict[str, str]
 ) -> types.Message:
-    driver = get_driver(ttl_hash=get_ttl_hash(seconds=10))
+    driver = get_driver()
     text = parse_site(
         driver=driver,
         url='https://yandex.ru/maps/213/moscow/stops/stop__9640740/'
@@ -53,7 +53,7 @@ async def home_office(
 async def office_home(
     query: types.CallbackQuery, callback_data: dict[str, str]
 ) -> types.Message:
-    driver = get_driver(ttl_hash=get_ttl_hash())
+    driver = get_driver()
     text = parse_site(
         driver=driver,
         url='https://yandex.ru/maps/213/moscow/stops/stop__9640288/?'
@@ -79,7 +79,7 @@ async def echo(message: types.Message) -> types.Message:
 
 
 async def morning_bus_mailing(chat_ids: list[int]) -> None:
-    driver = get_driver(ttl_hash=get_ttl_hash())
+    driver = get_driver()
     text = parse_site(
         driver=driver,
         url='https://yandex.ru/maps/213/moscow/stops/stop__9640740/'

app/core/logger.py

@@ -1,11 +0,0 @@
-import sys
-
-from loguru import logger
-
-logger.remove()
-logger.add(
-    sink=sys.stdout,
-    colorize=True,
-    level='DEBUG',
-    format="<cyan>{time:DD.MM.YYYY HH:mm:ss}</cyan> | <level>{level}</level> | <magenta>{message}</magenta>",
-)

app/core/parse_web.py

@@ -1,12 +1,11 @@
 import os
 import tarfile
 import time
-from functools import lru_cache
 from pathlib import Path
 
 import wget
-from app.core.logger import logger
-from app.settings import BASE_DIR, GECKO_DRIVER_VERSION
+from app.core.utils import logger, timed_cache
+from app.settings import BASE_DIR, DRIVER_SESSION_TTL, GECKO_DRIVER_VERSION
 from selenium import webdriver
 from selenium.common.exceptions import (
     NoSuchElementException,
@@ -93,14 +92,8 @@ def parse_site(url: str, message: str, driver: RemoteWebDriver | None = None) ->
     return answer
 
 
-def get_ttl_hash(seconds: int = 28) -> int:
-    """Return the same value withing `seconds` time period"""
-    return round(time.time() / seconds)
-
-
-@lru_cache
-def get_driver(ttl_hash: int | None = None) -> RemoteWebDriver:
-    del ttl_hash
+@timed_cache(seconds=DRIVER_SESSION_TTL)
+def get_driver() -> RemoteWebDriver:
     opt = options.Options()
     opt.headless = True
     driver = RemoteWebDriver(

app/core/utils.py

@@ -0,0 +1,35 @@
+import sys
+from datetime import datetime, timedelta
+from functools import lru_cache, wraps
+from typing import Any
+
+from loguru import logger
+
+logger.remove()
+logger.add(
+    sink=sys.stdout,
+    colorize=True,
+    level='DEBUG',
+    format="<cyan>{time:DD.MM.YYYY HH:mm:ss}</cyan> | <level>{level}</level> | <magenta>{message}</magenta>",
+)
+
+
+def timed_cache(**timedelta_kwargs: Any) -> Any:
+    def _wrapper(func: Any) -> Any:
+        update_delta = timedelta(**timedelta_kwargs)
+        next_update = datetime.utcnow() + update_delta
+        # Apply @lru_cache to f with no cache size limit
+        cached_func = lru_cache(None)(func)
+
+        @wraps(func)
+        def _wrapped(*args: Any, **kwargs: Any) -> Any:
+            nonlocal next_update
+            now = datetime.utcnow()
+            if now >= next_update:
+                cached_func.cache_clear()
+                next_update = now + update_delta
+            return cached_func(*args, **kwargs)
+
+        return _wrapped
+
+    return _wrapper