on 0 ttl no caching

Author: agkloop

README.md

@@ -88,16 +88,6 @@ user = await get_user_async(42)
 ```
 
 ## Benchmarks
-
-**Performance Comparison** (10ms baseline operation):
-
-| Strategy | Time | Speedup |
-|----------|------|---------|
-| No Cache | 12.51 ms | 1x |
-| TTLCache | 0.0010 ms | 12,000x  |
-| SWRCache | 0.0014 ms | 9,100x  |
-| BGCache | 0.0003 ms | 37,000x  |
-
 Full benchmarks available in `tests/benchmark.py`.
 
 ## API Reference

src/advanced_caching/decorators.py

@@ -20,10 +20,11 @@
 
 from .storage import InMemCache, CacheEntry, CacheStorage
 
-logger = logging.getLogger(__name__)
-
 T = TypeVar("T")
 
+# Minimal logger used only for error reporting (no debug/info on hot paths)
+logger = logging.getLogger(__name__)
+
 
 # ============================================================================
 # TTLCache - Simple TTL-based caching decorator
@@ -81,6 +82,9 @@ def calculate(x):
 
         def decorator(func: Callable[..., T]) -> Callable[..., T]:
             def wrapper(*args, **kwargs) -> T:
+                # If ttl is 0 or negative, disable caching and call through
+                if ttl <= 0:
+                    return func(*args, **kwargs)
                 # Generate cache key
                 if callable(key):
                     cache_key = key(*args, **kwargs)
@@ -178,6 +182,9 @@ def get_user(user_id: int):
 
         def decorator(func: Callable[..., T]) -> Callable[..., T]:
             def wrapper(*args, **kwargs) -> T:
+                # If ttl is 0 or negative, disable caching and SWR behavior
+                if ttl <= 0:
+                    return func(*args, **kwargs)
                 # Generate cache key
                 if callable(key):
                     # Key function is responsible for handling args/defaults
@@ -212,7 +219,6 @@ def wrapper(*args, **kwargs) -> T:
 
                 if entry is None:
                     # Cache miss - fetch now
-                    logger.debug(f"Cache MISS: {cache_key}")
                     result = func(*args, **kwargs)
                     now = time.time()
                     cache_entry = CacheEntry(
@@ -221,16 +227,12 @@ def wrapper(*args, **kwargs) -> T:
                     function_cache.set_entry(cache_key, cache_entry)
                     return result
 
-                # Check if fresh (within TTL)
                 if entry.is_fresh():
-                    logger.debug(f"Cache HIT (fresh): {cache_key}")
                     return entry.value
 
-                # Stale - check if still within stale period
                 age = entry.age()
                 if age > (ttl + stale_ttl):
                     # Too stale, fetch now
-                    logger.debug(f"Cache HIT (too stale): {cache_key}, age={age:.1f}s")
                     result = func(*args, **kwargs)
                     now = time.time()
                     cache_entry = CacheEntry(
@@ -240,10 +242,6 @@ def wrapper(*args, **kwargs) -> T:
                     return result
 
                 # Stale but within grace period - return stale and refresh in background
-                logger.debug(
-                    f"Cache HIT (stale): {cache_key}, refreshing in background"
-                )
-
                 # Try to acquire refresh lock
                 lock_key = f"{cache_key}:refresh_lock"
                 if enable_lock:
@@ -262,9 +260,11 @@ def refresh_job():
                             value=new_value, fresh_until=now + ttl, created_at=now
                         )
                         function_cache.set_entry(cache_key, cache_entry)
-                        logger.debug(f"Background refresh complete: {cache_key}")
-                    except Exception as e:
-                        logger.error(f"Background refresh failed for {cache_key}: {e}")
+                    except Exception:
+                        # Log background refresh failures but never raise
+                        logger.exception(
+                            "SWR background refresh failed for key %r", cache_key
+                        )
 
                 thread = threading.Thread(target=refresh_job, daemon=True)
                 thread.start()
@@ -315,7 +315,6 @@ def start(cls) -> None:
             if not cls._started:
                 cls.get_scheduler().start()
                 cls._started = True
-                logger.info("Shared BackgroundScheduler started")
 
     @classmethod
     def shutdown(cls, wait: bool = True) -> None:
@@ -325,7 +324,6 @@ def shutdown(cls, wait: bool = True) -> None:
                 cls._scheduler.shutdown(wait=wait)
                 cls._started = False
                 cls._scheduler = None
-                logger.info("Shared BackgroundScheduler stopped")
 
 
 # ============================================================================
@@ -390,8 +388,13 @@ def register_loader(
             Decorated function that returns cached data (sync or async).
         """
         cache_key = key
-        if ttl is None:
+        # If interval_seconds <= 0 or ttl == 0, disable background scheduling and caching.
+        if interval_seconds <= 0:
+            interval_seconds = 0
+        if ttl is None and interval_seconds > 0:
             ttl = interval_seconds * 2
+        if ttl is None:
+            ttl = 0
 
         # Create a dedicated cache instance for this loader
         loader_cache = cache if cache is not None else InMemCache()
@@ -400,16 +403,38 @@ def decorator(loader_func: Callable[[], T]) -> Callable[[], T]:
             # Detect if function is async
             is_async = asyncio.iscoroutinefunction(loader_func)
 
+            # If no scheduling/caching is desired, just wrap the function and call through
+            if interval_seconds <= 0 or ttl <= 0:
+                if is_async:
+
+                    async def async_wrapper() -> T:
+                        return await loader_func()
+
+                    async_wrapper.__wrapped__ = loader_func  # type: ignore
+                    async_wrapper.__name__ = loader_func.__name__  # type: ignore
+                    async_wrapper.__doc__ = loader_func.__doc__  # type: ignore
+                    async_wrapper._cache = loader_cache  # type: ignore
+                    async_wrapper._cache_key = cache_key  # type: ignore
+
+                    return async_wrapper  # type: ignore
+                else:
+
+                    def sync_wrapper() -> T:
+                        return loader_func()
+
+                    sync_wrapper.__wrapped__ = loader_func  # type: ignore
+                    sync_wrapper.__name__ = loader_func.__name__  # type: ignore
+                    sync_wrapper.__doc__ = loader_func.__doc__  # type: ignore
+                    sync_wrapper._cache = loader_cache  # type: ignore
+                    sync_wrapper._cache_key = cache_key  # type: ignore
+
+                    return sync_wrapper  # type: ignore
+
             # Create wrapper that loads and caches
             def refresh_job():
                 """Job that runs periodically to refresh the cache."""
                 try:
-                    logger.debug(f"Refreshing cache key: {cache_key}")
-                    start = time.time()
-
-                    # Call function (async or sync)
                     if is_async:
-                        # Run async function in new event loop
                         loop = asyncio.new_event_loop()
                         asyncio.set_event_loop(loop)
                         try:
@@ -419,19 +444,22 @@ def refresh_job():
                     else:
                         data = loader_func()
 
-                    duration = time.time() - start
-
                     loader_cache.set(cache_key, data, ttl)
-                    logger.info(
-                        f"Refreshed {cache_key} successfully in {duration:.3f}s"
-                    )
                 except Exception as e:
-                    logger.error(f"Failed to refresh {cache_key}: {e}", exc_info=True)
+                    # User-provided error handler gets first chance
                     if on_error:
                         try:
                             on_error(e)
-                        except Exception as err:
-                            logger.error(f"Error handler failed: {err}")
+                        except Exception:
+                            # Avoid user handler breaking the scheduler
+                            logger.exception(
+                                "BGCache error handler failed for key %r", cache_key
+                            )
+                    else:
+                        # Log uncaught loader errors for visibility
+                        logger.exception(
+                            "BGCache refresh job failed for key %r", cache_key
+                        )
 
             # Get shared scheduler
             scheduler = _SharedScheduler.get_scheduler()

tests/test_correctness.py

@@ -606,5 +606,65 @@ def f(*, x: int) -> int:
         assert calls["n"] == 1
 
 
+class TestNoCachingWhenZero:
+    """Ensure ttl/interval_seconds == 0 disables caching/background behavior."""
+
+    def test_ttlcache_ttl_zero_disables_caching(self):
+        calls = {"n": 0}
+
+        @TTLCache.cached("user:{}", ttl=0)
+        def get_user(user_id: int) -> int:
+            calls["n"] += 1
+            return calls["n"]
+
+        # Each call should invoke the function (no caching)
+        assert get_user(1) == 1
+        assert get_user(1) == 2
+        assert get_user(1) == 3
+        assert calls["n"] == 3
+
+    def test_swrcache_ttl_zero_disables_caching(self):
+        calls = {"n": 0}
+
+        @SWRCache.cached("data:{}", ttl=0, stale_ttl=10)
+        def get_data(key: str) -> int:
+            calls["n"] += 1
+            return calls["n"]
+
+        # Each call should invoke the function (no SWR behavior)
+        assert get_data("k") == 1
+        assert get_data("k") == 2
+        assert get_data("k") == 3
+        assert calls["n"] == 3
+
+    def test_bgcache_interval_zero_disables_background_and_cache(self):
+        calls = {"n": 0}
+
+        @BGCache.register_loader(key="no_bg", interval_seconds=0, ttl=None)
+        def load_data() -> int:
+            calls["n"] += 1
+            return calls["n"]
+
+        # No background scheduler, no caching: each call increments
+        assert load_data() == 1
+        assert load_data() == 2
+        assert load_data() == 3
+        assert calls["n"] == 3
+
+    def test_bgcache_ttl_zero_disables_background_and_cache(self):
+        calls = {"n": 0}
+
+        @BGCache.register_loader(key="no_bg_ttl", interval_seconds=10, ttl=0)
+        def load_data() -> int:
+            calls["n"] += 1
+            return calls["n"]
+
+        # Because ttl == 0, wrapper should bypass cache and scheduler
+        assert load_data() == 1
+        assert load_data() == 2
+        assert load_data() == 3
+        assert calls["n"] == 3
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])