refactor: Iran holiday TTL caching, callable TTL on ttl_cache_decorator, and ty/grpc cleanups

Author: HosseinNejatiJavaremi

archipy/adapters/email/adapters.py

@@ -2,7 +2,6 @@
 import logging
 import os
 import smtplib
-from collections.abc import Callable
 from datetime import datetime
 from email import encoders
 from email.mime.audio import MIMEAudio
@@ -12,7 +11,7 @@
 from email.mime.text import MIMEText
 from pathlib import Path
 from queue import Queue
-from typing import Any, BinaryIO, cast, override
+from typing import BinaryIO, override
 
 import requests
 from jinja2 import Template
@@ -162,8 +161,7 @@ def _process_source(source: str | bytes | BinaryIO | HttpUrl, attachment_type: E
             if hasattr(source, "read"):
                 read_method = source.read
                 if callable(read_method):
-                    read_callable = cast("Callable[[], Any]", read_method)
-                    result = read_callable()
+                    result = read_method()  # ty: ignore[call-top-callable]
                     if isinstance(result, bytes):
                         return result
                     if isinstance(result, str):

archipy/helpers/decorators/cache.py

@@ -110,23 +110,25 @@ def clear_cache(self) -> None:
 
 
 def ttl_cache_decorator[**P, R](
-    ttl_seconds: int = 300,
+    ttl_seconds: int | Callable[[], int] = 300,
     maxsize: int = 100,
-) -> Callable[[Callable[P, R]], CachedFunction[P, R]]:
+) -> Any:
     """Decorator that provides a TTL cache for functions and methods.
 
     The cache is shared across all instances when decorating instance methods.
     This is by design to allow efficient caching of expensive operations that
     depend only on the method arguments, not the instance state.
 
     Args:
-        ttl_seconds: Time to live in seconds (default: 5 minutes).
+        ttl_seconds: Time to live in seconds (default: 5 minutes), or a zero-argument
+            callable that returns the TTL (evaluated once when the decorator is applied).
             After this time, cached entries expire and the function is re-executed.
         maxsize: Maximum size of the cache (default: 100).
             When the cache is full, the least recently used entry is evicted.
 
     Returns:
-        Decorated function with TTL caching and a clear_cache() method.
+        Decorated function with TTL caching and a clear_cache() method (typed as :class:`~typing.Any`
+        at compile time so ParamSpec-style decorator typing stays sound).
 
     Example:
         ```python
@@ -157,9 +159,13 @@ def fetch_data(self, key: str) -> dict:
     """
     from cachetools import TTLCache
 
-    cache: TTLCache = TTLCache(maxsize=maxsize, ttl=ttl_seconds)
+    if isinstance(ttl_seconds, int):
+        resolved_ttl = int(ttl_seconds)
+    else:
+        resolved_ttl = int(ttl_seconds())
+    cache: TTLCache = TTLCache(maxsize=maxsize, ttl=resolved_ttl)
 
     def decorator(func: Callable[P, R]) -> CachedFunction[P, R]:
         return CachedFunction(func, cache, instance=None)
 
-    return decorator  # type: ignore[return-value]
+    return decorator

archipy/helpers/interceptors/grpc/metric/server_interceptor.py

@@ -1,7 +1,7 @@
 import asyncio
 import time
 from collections.abc import Callable
-from typing import Any, ClassVar, cast
+from typing import ClassVar
 
 import grpc
 
@@ -90,8 +90,8 @@ def intercept(
             result = method(request, context)
 
             if hasattr(context, "code") and callable(context.code):
-                code_method = cast("Callable[[], Any]", context.code)
-                code_obj = code_method()
+                code_method = context.code
+                code_obj = code_method()  # ty: ignore[call-top-callable]
                 if code_obj is not None:
                     code_name = getattr(code_obj, "name", None)
                     if code_name is not None:
@@ -198,8 +198,8 @@ async def intercept(
                         if code_name is not None:
                             status_code = code_name
                 elif hasattr(e, "code") and callable(e.code):
-                    code_method = cast("Callable[[], Any]", e.code)
-                    code_obj = code_method()
+                    code_method = e.code
+                    code_obj = code_method()  # ty: ignore[call-top-callable]
                     if code_obj is not None:
                         code_name = getattr(code_obj, "name", None)
                         if code_name is not None:

archipy/helpers/utils/datetime_utils.py

@@ -1,16 +1,29 @@
 import time
 from collections.abc import Generator
 from datetime import UTC, date, datetime, timedelta
-from typing import Any, ClassVar
+from typing import Any
 
 import jdatetime
 import requests
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
 from archipy.configs.base_config import BaseConfig
+from archipy.helpers.decorators.cache import ttl_cache_decorator
 from archipy.models.errors import UnknownError
 
+_HOLIDAY_CACHE_MAXSIZE = 2048
+
+
+def _iran_holiday_historical_cache_ttl_seconds() -> int:
+    """Return DATETIME.HISTORICAL_CACHE_TTL from the global config (evaluated at decorate time)."""
+    return BaseConfig.global_config().DATETIME.HISTORICAL_CACHE_TTL
+
+
+def _iran_holiday_standard_cache_ttl_seconds() -> int:
+    """Return DATETIME.CACHE_TTL from the global config (evaluated at decorate time)."""
+    return BaseConfig.global_config().DATETIME.CACHE_TTL
+
 
 class DatetimeUtils:
     """A utility class for handling date and time operations, including conversions, caching, and API integrations.
@@ -19,9 +32,6 @@ class DatetimeUtils:
     utility functions for timezone-aware datetime objects, date ranges, and string formatting.
     """
 
-    """A class-level cache for storing holiday statuses to avoid redundant API calls."""
-    _holiday_cache: ClassVar[dict[str, tuple[bool, datetime]]] = {}
-
     @staticmethod
     def convert_to_jalali(target_date: date) -> jdatetime.date:
         """Converts a Gregorian date to a Jalali (Persian) date.
@@ -46,83 +56,42 @@ def is_holiday_in_iran(cls, target_date: date) -> bool:
         Returns:
             bool: True if the date is a holiday, False otherwise.
         """
-        # Convert to Jalali date first
-        jalali_date = cls.convert_to_jalali(target_date)
         date_str = target_date.strftime("%Y-%m-%d")
-        current_time = cls.get_datetime_utc_now()
-
-        # Check cache first
-        is_cached, is_holiday = cls._check_cache(date_str, current_time)
-        if is_cached:
-            return is_holiday
-
-        # Fetch holiday status and cache it
-        return cls._fetch_and_cache_holiday_status(jalali_date, date_str, current_time)
-
-    @classmethod
-    def _check_cache(cls, date_str: str, current_time: datetime) -> tuple[bool, bool]:
-        """Checks the cache for holiday status to avoid redundant API calls.
-
-        Args:
-            date_str (str): The date string to check in the cache.
-            current_time (datetime): The current time to compare against cache expiration.
-
-        Returns:
-            tuple[bool, bool]: A tuple where the first element indicates if the cache was hit,
-                               and the second element is the cached holiday status.
-        """
-        cached_data = cls._holiday_cache.get(date_str)
-        if cached_data:
-            is_holiday, expiry_time = cached_data
-            if current_time < expiry_time:
-                return True, is_holiday
-
-            # Remove expired cache entry
-            del cls._holiday_cache[date_str]
-
-        return False, False
+        utc_today = cls.get_datetime_utc_now().date()
+        is_historical = target_date <= utc_today
+        if is_historical:
+            return cls._fetch_holiday_in_iran_historical(date_str)
+        return cls._fetch_holiday_in_iran_standard(date_str)
 
-    @classmethod
-    def _fetch_and_cache_holiday_status(
-        cls,
-        jalali_date: jdatetime.date,
-        date_str: str,
-        current_time: datetime,
-    ) -> bool:
-        """Fetches holiday status from the API and caches the result.
-
-        This method calls an external API to determine if the given Jalali date is a holiday.
-        If the API call is successful, the result is cached with an expiration time to avoid
-        redundant API calls. If the API call fails, an `UnknownError` is raised.
-
-        Args:
-            jalali_date (jdatetime.date): The Jalali date to check for holiday status.
-            date_str (str): The date string to use as a cache key.
-            current_time (datetime): The current time to set cache expiration.
-
-        Returns:
-            bool: True if the date is a holiday, False otherwise.
-
-        Raises:
-            UnknownError: If the API request fails due to a network issue or other request-related errors.
-        """
+    @staticmethod
+    @ttl_cache_decorator(
+        ttl_seconds=_iran_holiday_historical_cache_ttl_seconds,
+        maxsize=_HOLIDAY_CACHE_MAXSIZE,
+    )
+    def _fetch_holiday_in_iran_historical(date_str: str) -> bool:
+        """Resolve holiday flag for a date on or before UTC today; cached with historical TTL."""
+        target_date = datetime.strptime(date_str, "%Y-%m-%d").date()
+        jalali_date = DatetimeUtils.convert_to_jalali(target_date)
         try:
-            config: Any = BaseConfig.global_config()
-            response = cls._call_holiday_api(jalali_date)
-            is_holiday = cls._parse_holiday_response(response, jalali_date)
-
-            # Determine cache TTL based on whether the date is historical
-            target_date = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=UTC).date()
-            is_historical = target_date <= current_time.date()
-            cache_ttl = config.DATETIME.HISTORICAL_CACHE_TTL if is_historical else config.DATETIME.CACHE_TTL
-
-            # Cache the result with appropriate expiration
-            expiry_time = current_time + timedelta(seconds=cache_ttl)
-            cls._holiday_cache[date_str] = (is_holiday, expiry_time)
+            response = DatetimeUtils._call_holiday_api(jalali_date)
         except requests.RequestException as exception:
             raise UnknownError from exception
+        return DatetimeUtils._parse_holiday_response(response, jalali_date)
 
-        return is_holiday
+    @staticmethod
+    @ttl_cache_decorator(
+        ttl_seconds=_iran_holiday_standard_cache_ttl_seconds,
+        maxsize=_HOLIDAY_CACHE_MAXSIZE,
+    )
+    def _fetch_holiday_in_iran_standard(date_str: str) -> bool:
+        """Resolve holiday flag for a strictly future calendar date; cached with standard TTL."""
+        target_date = datetime.strptime(date_str, "%Y-%m-%d").date()
+        jalali_date = DatetimeUtils.convert_to_jalali(target_date)
+        try:
+            response = DatetimeUtils._call_holiday_api(jalali_date)
+        except requests.RequestException as exception:
+            raise UnknownError from exception
+        return DatetimeUtils._parse_holiday_response(response, jalali_date)
 
     @staticmethod
     def _call_holiday_api(jalali_date: jdatetime.date) -> dict[str, Any]:

archipy/helpers/utils/keycloak_utils.py

@@ -2,7 +2,7 @@
 import logging
 from collections.abc import Callable
 from contextvars import ContextVar
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from grpc import ServicerContext
@@ -47,6 +47,24 @@
 logger = logging.getLogger(__name__)
 
 
+def _abort_grpc_sync_if_servicer_context(error: BaseError, context: object) -> None:
+    """Invoke ``error.abort_grpc_sync`` when ``context`` is a sync gRPC servicer context."""
+    if not GRPC_AVAILABLE or not hasattr(error, "abort_grpc_sync"):
+        return
+    if isinstance(context, ServicerContext):
+        servicer_ctx = context
+        error.abort_grpc_sync(servicer_ctx)  # ty: ignore[invalid-argument-type]
+
+
+async def _abort_grpc_async_if_servicer_context(error: BaseError, context: object) -> None:
+    """Invoke ``error.abort_grpc_async`` when ``context`` is an async gRPC servicer context."""
+    if not GRPC_AVAILABLE or not hasattr(error, "abort_grpc_async"):
+        return
+    if isinstance(context, AsyncServicerContext):
+        aio_ctx = context
+        await error.abort_grpc_async(aio_ctx)  # ty: ignore[invalid-argument-type]
+
+
 class AuthContext(BaseModel):
     """Authentication context passed to business logic."""
 
@@ -312,10 +330,10 @@ async def dependency(
     @staticmethod
     def _extract_token_from_metadata(context: object) -> str | None:
         """Extract Bearer token from gRPC metadata."""
-        if not hasattr(context, "invocation_metadata") or not callable(context.invocation_metadata):
+        get_metadata = getattr(context, "invocation_metadata", None)
+        if get_metadata is None or not callable(get_metadata):
             return None
-        invocation_metadata_method = cast("Callable[[], Any]", context.invocation_metadata)
-        invocation_metadata_result = invocation_metadata_method()
+        invocation_metadata_result = get_metadata()
         if invocation_metadata_result is None:
             return None
         # Convert metadata tuples to dict, handling both str and bytes keys
@@ -478,10 +496,8 @@ def wrapper(self: object, request: object, context: object) -> object:
                     return func(self, request, context)
 
                 except Exception as e:
-                    if isinstance(e, BaseError) and hasattr(e, "abort_grpc_sync") and GRPC_AVAILABLE:
-                        # Only call abort if context is actually a ServicerContext
-                        if hasattr(context, "abort"):
-                            e.abort_grpc_sync(context)  # type: ignore[arg-type]
+                    if isinstance(e, BaseError):
+                        _abort_grpc_sync_if_servicer_context(e, context)
                     raise InternalError(
                         lang=lang,
                         additional_data={"original_error": str(e), "error_type": type(e).__name__},
@@ -623,19 +639,18 @@ async def wrapper(self: object, request: object, context: object) -> object:
                     return await func(self, request, context)
 
                 except Exception as e:
-                    if context is None:
+                    grpc_ctx = context
+                    if grpc_ctx is None:
                         raise
-                    if isinstance(e, BaseError) and GRPC_AVAILABLE:
-                        if isinstance(context, AsyncServicerContext):
-                            await e.abort_grpc_async(context)  # type: ignore[arg-type]
-                            return None  # abort_grpc_async will terminate, but satisfy type checker
-                    if GRPC_AVAILABLE and isinstance(context, AsyncServicerContext):
-                        # False positive: isinstance narrows type at runtime
+                    if isinstance(e, BaseError) and GRPC_AVAILABLE and isinstance(grpc_ctx, AsyncServicerContext):
+                        await _abort_grpc_async_if_servicer_context(e, grpc_ctx)
+                        return None  # abort_grpc_async will terminate, but satisfy type checker
+                    if GRPC_AVAILABLE and isinstance(grpc_ctx, AsyncServicerContext):
                         error_instance = InternalError(
                             lang=lang,
                             additional_data={"original_error": str(e), "error_type": type(e).__name__},
                         )
-                        await error_instance.abort_grpc_async(context)  # type: ignore[arg-type]
+                        await _abort_grpc_async_if_servicer_context(error_instance, grpc_ctx)
                         return None  # abort_grpc_async will terminate, but satisfy type checker
                     raise
 

archipy/helpers/utils/string_utils.py

@@ -323,11 +323,13 @@ def replace_numbers_with_mask(cls, text: str, mask: str | None = None) -> str:
             str: The text with numbers masked.
         """
         mask = mask or "MASK_NUMBERS"
-        numbers = re.findall("[0-9]+", text)
-        result: str = text
-        for number in sorted(numbers, key=len, reverse=True):
-            result = result.replace(number, f" {mask} ")  # type: ignore[arg-type]
-        return result
+        work_text = str(text)
+        numbers: list[str] = re.findall("[0-9]+", work_text)
+        replacement = f" {mask} "
+        for raw_number in sorted(numbers, key=len, reverse=True):
+            number = str(raw_number)
+            work_text = re.sub(re.escape(number), replacement, work_text)
+        return work_text
 
     @classmethod
     def is_string_none_or_empty(cls, text: str) -> bool:

archipy/models/dtos/range_dtos.py

@@ -45,7 +45,7 @@ def validate_range(self) -> Self:
             # Use comparison with proper type handling
             # The protocol ensures both values support comparison
             try:
-                if self.from_ > self.to:  # type: ignore[operator]
+                if self.from_ > self.to:  # ty: ignore[unsupported-operator]
                     raise OutOfRangeError(field_name="from_")
             except TypeError:
                 # If comparison fails, skip validation (shouldn't happen with proper types)

features/datetime_utils.feature

@@ -74,7 +74,7 @@ Feature: Datetime Utilities
     Then the result should be cached with historical TTL
 
   Scenario: Verify current dates use standard cache TTL
-#    Test that current/future dates get cached with standard CACHE_TTL
-    Given a current Gregorian date "2026-03-21"
+#    Test that strictly future calendar dates get standard CACHE_TTL (today/past use historical TTL).
+    Given a Gregorian date strictly after today
     When we check if the date is a holiday in Iran with cache verification
     Then the result should be cached with standard TTL