Skip to content

Cache Service

cache

Caching system for Weeb CLI.

This module provides a two-tier caching system with both memory and file-based storage. Supports TTL (time-to-live) for automatic cache expiration.

The cache system is used throughout the application to reduce redundant API calls and improve performance.

Classes:

Name Description
CacheManager

Main cache manager with memory and file storage

Functions:

Name Description
cached

Decorator for automatic function result caching
get_cache

Get global cache instance
Example

Using cache manager::

from weeb_cli.services.cache import get_cache

cache = get_cache()

# Store value
cache.set("search:animecix:naruto", results)

# Retrieve value (max 1 hour old)
results = cache.get("search:animecix:naruto", max_age=3600)

# Clear specific pattern
cache.clear_pattern("search:animecix:")

Using decorator::

from weeb_cli.services.cache import cached

@cached(max_age=1800)  # 30 minutes
def expensive_operation(param):
    # This result will be cached
    return compute_result(param)

CacheManager

Two-tier cache manager with memory and file storage.

Provides fast memory cache with persistent file backup. Automatically handles cache expiration based on TTL (time-to-live).

Attributes:

Name Type Description
cache_dir Path

Directory for cache file storage.
_memory_cache Dict[str, Tuple[Any, float]]

In-memory cache dictionary.
Source code in weeb_cli/services/cache.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class CacheManager:
    """Two-tier cache manager with memory and file storage.

    Provides fast memory cache with persistent file backup. Automatically
    handles cache expiration based on TTL (time-to-live).

    Attributes:
        cache_dir: Directory for cache file storage.
        _memory_cache: In-memory cache dictionary.
    """

    def __init__(self, cache_dir: Path) -> None:
        """Initialize cache manager.

        Args:
            cache_dir: Directory path for storing cache files.
        """
        self.cache_dir: Path = cache_dir
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self._memory_cache: Dict[str, Tuple[Any, float]] = {}

    def _get_cache_key(self, key: str) -> str:
        """Generate SHA256 hash for cache key.

        Args:
            key: Original cache key.

        Returns:
            Hexadecimal hash string.
        """
        return hashlib.sha256(key.encode()).hexdigest()

    def get(self, key: str, max_age: int = 3600) -> Optional[Any]:
        """Retrieve cached value if not expired.

        Checks memory cache first, then file cache. Automatically removes
        expired entries.

        Args:
            key: Cache key.
            max_age: Maximum age in seconds (default: 1 hour).

        Returns:
            Cached value if found and not expired, otherwise None.

        Example:
            >>> cache.get("search:naruto", max_age=1800)
            [{'id': '1', 'title': 'Naruto'}]
        """
        # Check memory cache
        if key in self._memory_cache:
            value, timestamp = self._memory_cache[key]
            if time.time() - timestamp < max_age:
                return value
            else:
                del self._memory_cache[key]

        # Check file cache
        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"

        if cache_file.exists():
            age = time.time() - cache_file.stat().st_mtime
            if age < max_age:
                try:
                    with open(cache_file, 'r', encoding='utf-8') as f:
                        value = json.load(f)
                        self._memory_cache[key] = (value, time.time())
                        return value
                except (json.JSONDecodeError, UnicodeDecodeError, OSError):
                    pass
                # Fallback: try reading as pickle for backward compatibility
                try:
                    with open(cache_file, 'rb') as f:
                        value = pickle.load(f)
                        self._memory_cache[key] = (value, time.time())
                        return value
                except (pickle.PickleError, EOFError, OSError):
                    cache_file.unlink(missing_ok=True)

        return None

    def set(self, key: str, value: Any) -> None:
        """Store value in cache.

        Stores in both memory and file cache for persistence.

        Args:
            key: Cache key.
            value: Value to cache (must be JSON-serializable).

        Example:
            >>> cache.set("search:naruto", results)
        """
        self._memory_cache[key] = (value, time.time())

        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"

        try:
            with open(cache_file, 'w', encoding='utf-8') as f:
                json.dump(value, f, ensure_ascii=False, default=str)
        except (OSError, TypeError):
            pass

    def delete(self, key: str) -> None:
        """Delete cached value.

        Removes from both memory and file cache.

        Args:
            key: Cache key to delete.
        """
        self._memory_cache.pop(key, None)

        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"
        cache_file.unlink(missing_ok=True)

    def clear(self) -> None:
        """Clear all cached values.

        Removes all entries from memory and deletes all cache files.
        """
        self._memory_cache.clear()

        for cache_file in self.cache_dir.glob("*.cache"):
            cache_file.unlink(missing_ok=True)

    def clear_pattern(self, pattern: str) -> int:
        """Clear cached values matching a pattern.

        Args:
            pattern: String pattern to match in cache keys.

        Returns:
            Number of entries removed.

        Example:
            >>> count = cache.clear_pattern("search:animecix:")
            >>> print(f"Removed {count} entries")
        """
        removed = 0
        keys_to_remove = [k for k in self._memory_cache.keys() if pattern in k]
        for key in keys_to_remove:
            del self._memory_cache[key]
            cache_key = self._get_cache_key(key)
            cache_file = self.cache_dir / f"{cache_key}.cache"
            cache_file.unlink(missing_ok=True)
            removed += 1

        return removed

    def invalidate_provider(self, provider_name: str) -> int:
        """Invalidate all cache entries for a provider.

        Clears search and details cache for the specified provider.

        Args:
            provider_name: Provider identifier.

        Returns:
            Number of entries removed.

        Example:
            >>> cache.invalidate_provider("animecix")
        """
        removed = 0
        patterns = [f"search:{provider_name}:", f"details:{provider_name}:"]

        for pattern in patterns:
            keys_to_remove = [k for k in list(self._memory_cache.keys()) if k.startswith(pattern)]
            for key in keys_to_remove:
                del self._memory_cache[key]
                removed += 1

        # Clear all file cache (conservative approach)
        for cache_file in self.cache_dir.glob("*.cache"):
            cache_file.unlink(missing_ok=True)

        return removed

    def cleanup(self, max_age: int = 86400) -> int:
        """Remove expired cache entries.

        Removes entries older than max_age from both memory and file cache.

        Args:
            max_age: Maximum age in seconds (default: 24 hours).

        Returns:
            Number of entries removed.

        Example:
            >>> # Remove entries older than 1 hour
            >>> count = cache.cleanup(max_age=3600)
        """
        removed = 0
        cutoff = time.time() - max_age

        # Clean memory cache
        keys_to_remove = []
        for key, (value, timestamp) in list(self._memory_cache.items()):
            if timestamp < cutoff:
                keys_to_remove.append(key)

        for key in keys_to_remove:
            del self._memory_cache[key]
            removed += 1

        # Clean file cache
        for cache_file in self.cache_dir.glob("*.cache"):
            if cache_file.stat().st_mtime < cutoff:
                cache_file.unlink(missing_ok=True)
                removed += 1

        return removed

    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics.

        Returns:
            Dictionary with cache statistics:
                - memory_entries: Number of in-memory entries
                - file_entries: Number of file cache entries
                - total_size_bytes: Total file cache size in bytes
                - total_size_mb: Total file cache size in MB

        Example:
            >>> stats = cache.get_stats()
            >>> print(f"Cache size: {stats['total_size_mb']} MB")
        """
        memory_count = len(self._memory_cache)
        file_count = len(list(self.cache_dir.glob("*.cache")))

        total_size = 0
        for cache_file in self.cache_dir.glob("*.cache"):
            total_size += cache_file.stat().st_size

        return {
            "memory_entries": memory_count,
            "file_entries": file_count,
            "total_size_bytes": total_size,
            "total_size_mb": round(total_size / (1024 * 1024), 2)
        }

__init__ 

__init__(cache_dir: Path) -> None

Initialize cache manager.

Parameters:

Name Type Description Default
cache_dir Path

Directory path for storing cache files.

 required
Source code in weeb_cli/services/cache.py 
62
63
64
65
66
67
68
69
70
def __init__(self, cache_dir: Path) -> None:
    """Initialize cache manager.

    Args:
        cache_dir: Directory path for storing cache files.
    """
    self.cache_dir: Path = cache_dir
    self.cache_dir.mkdir(parents=True, exist_ok=True)
    self._memory_cache: Dict[str, Tuple[Any, float]] = {}

get 

get(key: str, max_age: int = 3600) -> Optional[Any]

Retrieve cached value if not expired.

Checks memory cache first, then file cache. Automatically removes expired entries.

Parameters:

Name Type Description Default
key str

Cache key.

 required
max_age int

Maximum age in seconds (default: 1 hour).

 3600

Returns:

Type Description
Optional[Any]

Cached value if found and not expired, otherwise None.
Example

cache.get("search:naruto", max_age=1800) [{'id': '1', 'title': 'Naruto'}]

Source code in weeb_cli/services/cache.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get(self, key: str, max_age: int = 3600) -> Optional[Any]:
    """Retrieve cached value if not expired.

    Checks memory cache first, then file cache. Automatically removes
    expired entries.

    Args:
        key: Cache key.
        max_age: Maximum age in seconds (default: 1 hour).

    Returns:
        Cached value if found and not expired, otherwise None.

    Example:
        >>> cache.get("search:naruto", max_age=1800)
        [{'id': '1', 'title': 'Naruto'}]
    """
    # Check memory cache
    if key in self._memory_cache:
        value, timestamp = self._memory_cache[key]
        if time.time() - timestamp < max_age:
            return value
        else:
            del self._memory_cache[key]

    # Check file cache
    cache_key = self._get_cache_key(key)
    cache_file = self.cache_dir / f"{cache_key}.cache"

    if cache_file.exists():
        age = time.time() - cache_file.stat().st_mtime
        if age < max_age:
            try:
                with open(cache_file, 'r', encoding='utf-8') as f:
                    value = json.load(f)
                    self._memory_cache[key] = (value, time.time())
                    return value
            except (json.JSONDecodeError, UnicodeDecodeError, OSError):
                pass
            # Fallback: try reading as pickle for backward compatibility
            try:
                with open(cache_file, 'rb') as f:
                    value = pickle.load(f)
                    self._memory_cache[key] = (value, time.time())
                    return value
            except (pickle.PickleError, EOFError, OSError):
                cache_file.unlink(missing_ok=True)

    return None

set 

set(key: str, value: Any) -> None

Store value in cache.

Stores in both memory and file cache for persistence.

Parameters:

Name Type Description Default
key str

Cache key.

 required
value Any

Value to cache (must be JSON-serializable).

 required
Example

cache.set("search:naruto", results)

Source code in weeb_cli/services/cache.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def set(self, key: str, value: Any) -> None:
    """Store value in cache.

    Stores in both memory and file cache for persistence.

    Args:
        key: Cache key.
        value: Value to cache (must be JSON-serializable).

    Example:
        >>> cache.set("search:naruto", results)
    """
    self._memory_cache[key] = (value, time.time())

    cache_key = self._get_cache_key(key)
    cache_file = self.cache_dir / f"{cache_key}.cache"

    try:
        with open(cache_file, 'w', encoding='utf-8') as f:
            json.dump(value, f, ensure_ascii=False, default=str)
    except (OSError, TypeError):
        pass

delete 

delete(key: str) -> None

Delete cached value.

Removes from both memory and file cache.

Parameters:

Name Type Description Default
key str

Cache key to delete.

 required
Source code in weeb_cli/services/cache.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
def delete(self, key: str) -> None:
    """Delete cached value.

    Removes from both memory and file cache.

    Args:
        key: Cache key to delete.
    """
    self._memory_cache.pop(key, None)

    cache_key = self._get_cache_key(key)
    cache_file = self.cache_dir / f"{cache_key}.cache"
    cache_file.unlink(missing_ok=True)

clear 

clear() -> None

Clear all cached values.

Removes all entries from memory and deletes all cache files.

Source code in weeb_cli/services/cache.py 
170
171
172
173
174
175
176
177
178
def clear(self) -> None:
    """Clear all cached values.

    Removes all entries from memory and deletes all cache files.
    """
    self._memory_cache.clear()

    for cache_file in self.cache_dir.glob("*.cache"):
        cache_file.unlink(missing_ok=True)

clear_pattern 

clear_pattern(pattern: str) -> int

Clear cached values matching a pattern.

Parameters:

Name Type Description Default
pattern str

String pattern to match in cache keys.

 required

Returns:

Type Description
int

Number of entries removed.
Example

count = cache.clear_pattern("search:animecix:") print(f"Removed {count} entries")

Source code in weeb_cli/services/cache.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def clear_pattern(self, pattern: str) -> int:
    """Clear cached values matching a pattern.

    Args:
        pattern: String pattern to match in cache keys.

    Returns:
        Number of entries removed.

    Example:
        >>> count = cache.clear_pattern("search:animecix:")
        >>> print(f"Removed {count} entries")
    """
    removed = 0
    keys_to_remove = [k for k in self._memory_cache.keys() if pattern in k]
    for key in keys_to_remove:
        del self._memory_cache[key]
        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"
        cache_file.unlink(missing_ok=True)
        removed += 1

    return removed

invalidate_provider 

invalidate_provider(provider_name: str) -> int

Invalidate all cache entries for a provider.

Clears search and details cache for the specified provider.

Parameters:

Name Type Description Default
provider_name str

Provider identifier.

 required

Returns:

Type Description
int

Number of entries removed.
Example

cache.invalidate_provider("animecix")

Source code in weeb_cli/services/cache.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def invalidate_provider(self, provider_name: str) -> int:
    """Invalidate all cache entries for a provider.

    Clears search and details cache for the specified provider.

    Args:
        provider_name: Provider identifier.

    Returns:
        Number of entries removed.

    Example:
        >>> cache.invalidate_provider("animecix")
    """
    removed = 0
    patterns = [f"search:{provider_name}:", f"details:{provider_name}:"]

    for pattern in patterns:
        keys_to_remove = [k for k in list(self._memory_cache.keys()) if k.startswith(pattern)]
        for key in keys_to_remove:
            del self._memory_cache[key]
            removed += 1

    # Clear all file cache (conservative approach)
    for cache_file in self.cache_dir.glob("*.cache"):
        cache_file.unlink(missing_ok=True)

    return removed

cleanup 

cleanup(max_age: int = 86400) -> int

Remove expired cache entries.

Removes entries older than max_age from both memory and file cache.

Parameters:

Name Type Description Default
max_age int

Maximum age in seconds (default: 24 hours).

 86400

Returns:

Type Description
int

Number of entries removed.
Example
Remove entries older than 1 hour

count = cache.cleanup(max_age=3600)

Source code in weeb_cli/services/cache.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def cleanup(self, max_age: int = 86400) -> int:
    """Remove expired cache entries.

    Removes entries older than max_age from both memory and file cache.

    Args:
        max_age: Maximum age in seconds (default: 24 hours).

    Returns:
        Number of entries removed.

    Example:
        >>> # Remove entries older than 1 hour
        >>> count = cache.cleanup(max_age=3600)
    """
    removed = 0
    cutoff = time.time() - max_age

    # Clean memory cache
    keys_to_remove = []
    for key, (value, timestamp) in list(self._memory_cache.items()):
        if timestamp < cutoff:
            keys_to_remove.append(key)

    for key in keys_to_remove:
        del self._memory_cache[key]
        removed += 1

    # Clean file cache
    for cache_file in self.cache_dir.glob("*.cache"):
        if cache_file.stat().st_mtime < cutoff:
            cache_file.unlink(missing_ok=True)
            removed += 1

    return removed

get_stats 

get_stats() -> Dict[str, Any]

Get cache statistics.

Returns:

Type Description
Dict[str, Any]

Dictionary with cache statistics: - memory_entries: Number of in-memory entries - file_entries: Number of file cache entries - total_size_bytes: Total file cache size in bytes - total_size_mb: Total file cache size in MB
Example

stats = cache.get_stats() print(f"Cache size: {stats['total_size_mb']} MB")

Source code in weeb_cli/services/cache.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def get_stats(self) -> Dict[str, Any]:
    """Get cache statistics.

    Returns:
        Dictionary with cache statistics:
            - memory_entries: Number of in-memory entries
            - file_entries: Number of file cache entries
            - total_size_bytes: Total file cache size in bytes
            - total_size_mb: Total file cache size in MB

    Example:
        >>> stats = cache.get_stats()
        >>> print(f"Cache size: {stats['total_size_mb']} MB")
    """
    memory_count = len(self._memory_cache)
    file_count = len(list(self.cache_dir.glob("*.cache")))

    total_size = 0
    for cache_file in self.cache_dir.glob("*.cache"):
        total_size += cache_file.stat().st_size

    return {
        "memory_entries": memory_count,
        "file_entries": file_count,
        "total_size_bytes": total_size,
        "total_size_mb": round(total_size / (1024 * 1024), 2)
    }

cached 

cached(max_age: int = 3600, cache_manager: Optional[CacheManager] = None)

Decorator for automatic function result caching.

Caches function results based on arguments. Cache key is generated from function name and arguments.

Parameters:

Name Type Description Default
max_age int

Cache TTL in seconds (default: 1 hour).

 3600
cache_manager Optional[CacheManager]

Optional custom cache manager instance.

 None

Returns:

Type Description

Decorator function.
Example

@cached(max_age=1800) ... def fetch_anime_details(anime_id): ... # Expensive operation ... return api.get_details(anime_id)

First call: executes function

details = fetch_anime_details("123")

Second call within 30 min: returns cached result

details = fetch_anime_details("123")

Source code in weeb_cli/services/cache.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def cached(max_age: int = 3600, cache_manager: Optional[CacheManager] = None):
    """Decorator for automatic function result caching.

    Caches function results based on arguments. Cache key is generated
    from function name and arguments.

    Args:
        max_age: Cache TTL in seconds (default: 1 hour).
        cache_manager: Optional custom cache manager instance.

    Returns:
        Decorator function.

    Example:
        >>> @cached(max_age=1800)
        ... def fetch_anime_details(anime_id):
        ...     # Expensive operation
        ...     return api.get_details(anime_id)

        >>> # First call: executes function
        >>> details = fetch_anime_details("123")

        >>> # Second call within 30 min: returns cached result
        >>> details = fetch_anime_details("123")
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Generate cache key from function name and arguments
            key_parts = [func.__name__]
            key_parts.extend(str(arg) for arg in args)
            key_parts.extend(f"{k}={v}" for k, v in sorted(kwargs.items()))
            cache_key = ":".join(key_parts)

            cm = cache_manager or _get_global_cache()

            # Try to get from cache
            result = cm.get(cache_key, max_age)
            if result is not None:
                return result

            # Execute function and cache result
            result = func(*args, **kwargs)
            cm.set(cache_key, result)
            return result

        return wrapper
    return decorator

get_cache 

get_cache() -> CacheManager

Get global cache instance.

Returns:

Type Description
CacheManager

Global CacheManager instance for application-wide use.
Example

from weeb_cli.services.cache import get_cache cache = get_cache() cache.set("key", "value")

Source code in weeb_cli/services/cache.py 
364
365
366
367
368
369
370
371
372
373
374
375
def get_cache() -> CacheManager:
    """Get global cache instance.

    Returns:
        Global CacheManager instance for application-wide use.

    Example:
        >>> from weeb_cli.services.cache import get_cache
        >>> cache = get_cache()
        >>> cache.set("key", "value")
    """
    return _get_global_cache()

Overview

Two-tier caching system with memory and file-based storage for improved performance.

CacheManager

Main cache manager class.

Methods

  • get(): Retrieve cached value
  • set(): Store value in cache
  • delete(): Remove cached value
  • clear(): Clear all cache
  • clear_pattern(): Clear by pattern
  • invalidate_provider(): Clear provider cache
  • cleanup(): Remove expired entries
  • get_stats(): Get cache statistics

Usage Examples

Basic Caching

from weeb_cli.services.cache import get_cache

cache = get_cache()

# Store
cache.set("key", {"data": "value"})

# Retrieve
data = cache.get("key", max_age=3600)

Using Decorator

from weeb_cli.services.cache import cached

@cached(max_age=1800)
def expensive_function(param):
    # Result cached for 30 minutes
    return compute_result(param)

API Reference

CacheManager

Two-tier cache manager with memory and file storage.

Provides fast memory cache with persistent file backup. Automatically handles cache expiration based on TTL (time-to-live).

Attributes:

Name Type Description
cache_dir Path

Directory for cache file storage.
_memory_cache Dict[str, Tuple[Any, float]]

In-memory cache dictionary.
Source code in weeb_cli/services/cache.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class CacheManager:
    """Two-tier cache manager with memory and file storage.

    Provides fast memory cache with persistent file backup. Automatically
    handles cache expiration based on TTL (time-to-live).

    Attributes:
        cache_dir: Directory for cache file storage.
        _memory_cache: In-memory cache dictionary.
    """

    def __init__(self, cache_dir: Path) -> None:
        """Initialize cache manager.

        Args:
            cache_dir: Directory path for storing cache files.
        """
        self.cache_dir: Path = cache_dir
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self._memory_cache: Dict[str, Tuple[Any, float]] = {}

    def _get_cache_key(self, key: str) -> str:
        """Generate SHA256 hash for cache key.

        Args:
            key: Original cache key.

        Returns:
            Hexadecimal hash string.
        """
        return hashlib.sha256(key.encode()).hexdigest()

    def get(self, key: str, max_age: int = 3600) -> Optional[Any]:
        """Retrieve cached value if not expired.

        Checks memory cache first, then file cache. Automatically removes
        expired entries.

        Args:
            key: Cache key.
            max_age: Maximum age in seconds (default: 1 hour).

        Returns:
            Cached value if found and not expired, otherwise None.

        Example:
            >>> cache.get("search:naruto", max_age=1800)
            [{'id': '1', 'title': 'Naruto'}]
        """
        # Check memory cache
        if key in self._memory_cache:
            value, timestamp = self._memory_cache[key]
            if time.time() - timestamp < max_age:
                return value
            else:
                del self._memory_cache[key]

        # Check file cache
        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"

        if cache_file.exists():
            age = time.time() - cache_file.stat().st_mtime
            if age < max_age:
                try:
                    with open(cache_file, 'r', encoding='utf-8') as f:
                        value = json.load(f)
                        self._memory_cache[key] = (value, time.time())
                        return value
                except (json.JSONDecodeError, UnicodeDecodeError, OSError):
                    pass
                # Fallback: try reading as pickle for backward compatibility
                try:
                    with open(cache_file, 'rb') as f:
                        value = pickle.load(f)
                        self._memory_cache[key] = (value, time.time())
                        return value
                except (pickle.PickleError, EOFError, OSError):
                    cache_file.unlink(missing_ok=True)

        return None

    def set(self, key: str, value: Any) -> None:
        """Store value in cache.

        Stores in both memory and file cache for persistence.

        Args:
            key: Cache key.
            value: Value to cache (must be JSON-serializable).

        Example:
            >>> cache.set("search:naruto", results)
        """
        self._memory_cache[key] = (value, time.time())

        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"

        try:
            with open(cache_file, 'w', encoding='utf-8') as f:
                json.dump(value, f, ensure_ascii=False, default=str)
        except (OSError, TypeError):
            pass

    def delete(self, key: str) -> None:
        """Delete cached value.

        Removes from both memory and file cache.

        Args:
            key: Cache key to delete.
        """
        self._memory_cache.pop(key, None)

        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"
        cache_file.unlink(missing_ok=True)

    def clear(self) -> None:
        """Clear all cached values.

        Removes all entries from memory and deletes all cache files.
        """
        self._memory_cache.clear()

        for cache_file in self.cache_dir.glob("*.cache"):
            cache_file.unlink(missing_ok=True)

    def clear_pattern(self, pattern: str) -> int:
        """Clear cached values matching a pattern.

        Args:
            pattern: String pattern to match in cache keys.

        Returns:
            Number of entries removed.

        Example:
            >>> count = cache.clear_pattern("search:animecix:")
            >>> print(f"Removed {count} entries")
        """
        removed = 0
        keys_to_remove = [k for k in self._memory_cache.keys() if pattern in k]
        for key in keys_to_remove:
            del self._memory_cache[key]
            cache_key = self._get_cache_key(key)
            cache_file = self.cache_dir / f"{cache_key}.cache"
            cache_file.unlink(missing_ok=True)
            removed += 1

        return removed

    def invalidate_provider(self, provider_name: str) -> int:
        """Invalidate all cache entries for a provider.

        Clears search and details cache for the specified provider.

        Args:
            provider_name: Provider identifier.

        Returns:
            Number of entries removed.

        Example:
            >>> cache.invalidate_provider("animecix")
        """
        removed = 0
        patterns = [f"search:{provider_name}:", f"details:{provider_name}:"]

        for pattern in patterns:
            keys_to_remove = [k for k in list(self._memory_cache.keys()) if k.startswith(pattern)]
            for key in keys_to_remove:
                del self._memory_cache[key]
                removed += 1

        # Clear all file cache (conservative approach)
        for cache_file in self.cache_dir.glob("*.cache"):
            cache_file.unlink(missing_ok=True)

        return removed

    def cleanup(self, max_age: int = 86400) -> int:
        """Remove expired cache entries.

        Removes entries older than max_age from both memory and file cache.

        Args:
            max_age: Maximum age in seconds (default: 24 hours).

        Returns:
            Number of entries removed.

        Example:
            >>> # Remove entries older than 1 hour
            >>> count = cache.cleanup(max_age=3600)
        """
        removed = 0
        cutoff = time.time() - max_age

        # Clean memory cache
        keys_to_remove = []
        for key, (value, timestamp) in list(self._memory_cache.items()):
            if timestamp < cutoff:
                keys_to_remove.append(key)

        for key in keys_to_remove:
            del self._memory_cache[key]
            removed += 1

        # Clean file cache
        for cache_file in self.cache_dir.glob("*.cache"):
            if cache_file.stat().st_mtime < cutoff:
                cache_file.unlink(missing_ok=True)
                removed += 1

        return removed

    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics.

        Returns:
            Dictionary with cache statistics:
                - memory_entries: Number of in-memory entries
                - file_entries: Number of file cache entries
                - total_size_bytes: Total file cache size in bytes
                - total_size_mb: Total file cache size in MB

        Example:
            >>> stats = cache.get_stats()
            >>> print(f"Cache size: {stats['total_size_mb']} MB")
        """
        memory_count = len(self._memory_cache)
        file_count = len(list(self.cache_dir.glob("*.cache")))

        total_size = 0
        for cache_file in self.cache_dir.glob("*.cache"):
            total_size += cache_file.stat().st_size

        return {
            "memory_entries": memory_count,
            "file_entries": file_count,
            "total_size_bytes": total_size,
            "total_size_mb": round(total_size / (1024 * 1024), 2)
        }

__init__ 

__init__(cache_dir: Path) -> None

Initialize cache manager.

Parameters:

Name Type Description Default
cache_dir Path

Directory path for storing cache files.

 required
Source code in weeb_cli/services/cache.py 
62
63
64
65
66
67
68
69
70
def __init__(self, cache_dir: Path) -> None:
    """Initialize cache manager.

    Args:
        cache_dir: Directory path for storing cache files.
    """
    self.cache_dir: Path = cache_dir
    self.cache_dir.mkdir(parents=True, exist_ok=True)
    self._memory_cache: Dict[str, Tuple[Any, float]] = {}

get 

get(key: str, max_age: int = 3600) -> Optional[Any]

Retrieve cached value if not expired.

Checks memory cache first, then file cache. Automatically removes expired entries.

Parameters:

Name Type Description Default
key str

Cache key.

 required
max_age int

Maximum age in seconds (default: 1 hour).

 3600

Returns:

Type Description
Optional[Any]

Cached value if found and not expired, otherwise None.
Example

cache.get("search:naruto", max_age=1800) [{'id': '1', 'title': 'Naruto'}]

Source code in weeb_cli/services/cache.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get(self, key: str, max_age: int = 3600) -> Optional[Any]:
    """Retrieve cached value if not expired.

    Checks memory cache first, then file cache. Automatically removes
    expired entries.

    Args:
        key: Cache key.
        max_age: Maximum age in seconds (default: 1 hour).

    Returns:
        Cached value if found and not expired, otherwise None.

    Example:
        >>> cache.get("search:naruto", max_age=1800)
        [{'id': '1', 'title': 'Naruto'}]
    """
    # Check memory cache
    if key in self._memory_cache:
        value, timestamp = self._memory_cache[key]
        if time.time() - timestamp < max_age:
            return value
        else:
            del self._memory_cache[key]

    # Check file cache
    cache_key = self._get_cache_key(key)
    cache_file = self.cache_dir / f"{cache_key}.cache"

    if cache_file.exists():
        age = time.time() - cache_file.stat().st_mtime
        if age < max_age:
            try:
                with open(cache_file, 'r', encoding='utf-8') as f:
                    value = json.load(f)
                    self._memory_cache[key] = (value, time.time())
                    return value
            except (json.JSONDecodeError, UnicodeDecodeError, OSError):
                pass
            # Fallback: try reading as pickle for backward compatibility
            try:
                with open(cache_file, 'rb') as f:
                    value = pickle.load(f)
                    self._memory_cache[key] = (value, time.time())
                    return value
            except (pickle.PickleError, EOFError, OSError):
                cache_file.unlink(missing_ok=True)

    return None

set 

set(key: str, value: Any) -> None

Store value in cache.

Stores in both memory and file cache for persistence.

Parameters:

Name Type Description Default
key str

Cache key.

 required
value Any

Value to cache (must be JSON-serializable).

 required
Example

cache.set("search:naruto", results)

Source code in weeb_cli/services/cache.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def set(self, key: str, value: Any) -> None:
    """Store value in cache.

    Stores in both memory and file cache for persistence.

    Args:
        key: Cache key.
        value: Value to cache (must be JSON-serializable).

    Example:
        >>> cache.set("search:naruto", results)
    """
    self._memory_cache[key] = (value, time.time())

    cache_key = self._get_cache_key(key)
    cache_file = self.cache_dir / f"{cache_key}.cache"

    try:
        with open(cache_file, 'w', encoding='utf-8') as f:
            json.dump(value, f, ensure_ascii=False, default=str)
    except (OSError, TypeError):
        pass

delete 

delete(key: str) -> None

Delete cached value.

Removes from both memory and file cache.

Parameters:

Name Type Description Default
key str

Cache key to delete.

 required
Source code in weeb_cli/services/cache.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
def delete(self, key: str) -> None:
    """Delete cached value.

    Removes from both memory and file cache.

    Args:
        key: Cache key to delete.
    """
    self._memory_cache.pop(key, None)

    cache_key = self._get_cache_key(key)
    cache_file = self.cache_dir / f"{cache_key}.cache"
    cache_file.unlink(missing_ok=True)

clear 

clear() -> None

Clear all cached values.

Removes all entries from memory and deletes all cache files.

Source code in weeb_cli/services/cache.py 
170
171
172
173
174
175
176
177
178
def clear(self) -> None:
    """Clear all cached values.

    Removes all entries from memory and deletes all cache files.
    """
    self._memory_cache.clear()

    for cache_file in self.cache_dir.glob("*.cache"):
        cache_file.unlink(missing_ok=True)

clear_pattern 

clear_pattern(pattern: str) -> int

Clear cached values matching a pattern.

Parameters:

Name Type Description Default
pattern str

String pattern to match in cache keys.

 required

Returns:

Type Description
int

Number of entries removed.
Example

count = cache.clear_pattern("search:animecix:") print(f"Removed {count} entries")

Source code in weeb_cli/services/cache.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def clear_pattern(self, pattern: str) -> int:
    """Clear cached values matching a pattern.

    Args:
        pattern: String pattern to match in cache keys.

    Returns:
        Number of entries removed.

    Example:
        >>> count = cache.clear_pattern("search:animecix:")
        >>> print(f"Removed {count} entries")
    """
    removed = 0
    keys_to_remove = [k for k in self._memory_cache.keys() if pattern in k]
    for key in keys_to_remove:
        del self._memory_cache[key]
        cache_key = self._get_cache_key(key)
        cache_file = self.cache_dir / f"{cache_key}.cache"
        cache_file.unlink(missing_ok=True)
        removed += 1

    return removed

invalidate_provider 

invalidate_provider(provider_name: str) -> int

Invalidate all cache entries for a provider.

Clears search and details cache for the specified provider.

Parameters:

Name Type Description Default
provider_name str

Provider identifier.

 required

Returns:

Type Description
int

Number of entries removed.
Example

cache.invalidate_provider("animecix")

Source code in weeb_cli/services/cache.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def invalidate_provider(self, provider_name: str) -> int:
    """Invalidate all cache entries for a provider.

    Clears search and details cache for the specified provider.

    Args:
        provider_name: Provider identifier.

    Returns:
        Number of entries removed.

    Example:
        >>> cache.invalidate_provider("animecix")
    """
    removed = 0
    patterns = [f"search:{provider_name}:", f"details:{provider_name}:"]

    for pattern in patterns:
        keys_to_remove = [k for k in list(self._memory_cache.keys()) if k.startswith(pattern)]
        for key in keys_to_remove:
            del self._memory_cache[key]
            removed += 1

    # Clear all file cache (conservative approach)
    for cache_file in self.cache_dir.glob("*.cache"):
        cache_file.unlink(missing_ok=True)

    return removed

cleanup 

cleanup(max_age: int = 86400) -> int

Remove expired cache entries.

Removes entries older than max_age from both memory and file cache.

Parameters:

Name Type Description Default
max_age int

Maximum age in seconds (default: 24 hours).

 86400

Returns:

Type Description
int

Number of entries removed.
Example

Remove entries older than 1 hour

count = cache.cleanup(max_age=3600)

Source code in weeb_cli/services/cache.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def cleanup(self, max_age: int = 86400) -> int:
    """Remove expired cache entries.

    Removes entries older than max_age from both memory and file cache.

    Args:
        max_age: Maximum age in seconds (default: 24 hours).

    Returns:
        Number of entries removed.

    Example:
        >>> # Remove entries older than 1 hour
        >>> count = cache.cleanup(max_age=3600)
    """
    removed = 0
    cutoff = time.time() - max_age

    # Clean memory cache
    keys_to_remove = []
    for key, (value, timestamp) in list(self._memory_cache.items()):
        if timestamp < cutoff:
            keys_to_remove.append(key)

    for key in keys_to_remove:
        del self._memory_cache[key]
        removed += 1

    # Clean file cache
    for cache_file in self.cache_dir.glob("*.cache"):
        if cache_file.stat().st_mtime < cutoff:
            cache_file.unlink(missing_ok=True)
            removed += 1

    return removed

get_stats 

get_stats() -> Dict[str, Any]

Get cache statistics.

Returns:

Type Description
Dict[str, Any]

Dictionary with cache statistics: - memory_entries: Number of in-memory entries - file_entries: Number of file cache entries - total_size_bytes: Total file cache size in bytes - total_size_mb: Total file cache size in MB
Example

stats = cache.get_stats() print(f"Cache size: {stats['total_size_mb']} MB")

Source code in weeb_cli/services/cache.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def get_stats(self) -> Dict[str, Any]:
    """Get cache statistics.

    Returns:
        Dictionary with cache statistics:
            - memory_entries: Number of in-memory entries
            - file_entries: Number of file cache entries
            - total_size_bytes: Total file cache size in bytes
            - total_size_mb: Total file cache size in MB

    Example:
        >>> stats = cache.get_stats()
        >>> print(f"Cache size: {stats['total_size_mb']} MB")
    """
    memory_count = len(self._memory_cache)
    file_count = len(list(self.cache_dir.glob("*.cache")))

    total_size = 0
    for cache_file in self.cache_dir.glob("*.cache"):
        total_size += cache_file.stat().st_size

    return {
        "memory_entries": memory_count,
        "file_entries": file_count,
        "total_size_bytes": total_size,
        "total_size_mb": round(total_size / (1024 * 1024), 2)
    }

cached 

cached(max_age: int = 3600, cache_manager: Optional[CacheManager] = None)

Decorator for automatic function result caching.

Caches function results based on arguments. Cache key is generated from function name and arguments.

Parameters:

Name Type Description Default
max_age int

Cache TTL in seconds (default: 1 hour).

 3600
cache_manager Optional[CacheManager]

Optional custom cache manager instance.

 None

Returns:

Type Description

Decorator function.
Example

@cached(max_age=1800) ... def fetch_anime_details(anime_id): ... # Expensive operation ... return api.get_details(anime_id)

First call: executes function

details = fetch_anime_details("123")

Second call within 30 min: returns cached result

details = fetch_anime_details("123")

Source code in weeb_cli/services/cache.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def cached(max_age: int = 3600, cache_manager: Optional[CacheManager] = None):
    """Decorator for automatic function result caching.

    Caches function results based on arguments. Cache key is generated
    from function name and arguments.

    Args:
        max_age: Cache TTL in seconds (default: 1 hour).
        cache_manager: Optional custom cache manager instance.

    Returns:
        Decorator function.

    Example:
        >>> @cached(max_age=1800)
        ... def fetch_anime_details(anime_id):
        ...     # Expensive operation
        ...     return api.get_details(anime_id)

        >>> # First call: executes function
        >>> details = fetch_anime_details("123")

        >>> # Second call within 30 min: returns cached result
        >>> details = fetch_anime_details("123")
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Generate cache key from function name and arguments
            key_parts = [func.__name__]
            key_parts.extend(str(arg) for arg in args)
            key_parts.extend(f"{k}={v}" for k, v in sorted(kwargs.items()))
            cache_key = ":".join(key_parts)

            cm = cache_manager or _get_global_cache()

            # Try to get from cache
            result = cm.get(cache_key, max_age)
            if result is not None:
                return result

            # Execute function and cache result
            result = func(*args, **kwargs)
            cm.set(cache_key, result)
            return result

        return wrapper
    return decorator

get_cache 

get_cache() -> CacheManager

Get global cache instance.

Returns:

Type Description
CacheManager

Global CacheManager instance for application-wide use.
Example

from weeb_cli.services.cache import get_cache cache = get_cache() cache.set("key", "value")

Source code in weeb_cli/services/cache.py 
364
365
366
367
368
369
370
371
372
373
374
375
def get_cache() -> CacheManager:
    """Get global cache instance.

    Returns:
        Global CacheManager instance for application-wide use.

    Example:
        >>> from weeb_cli.services.cache import get_cache
        >>> cache = get_cache()
        >>> cache.set("key", "value")
    """
    return _get_global_cache()