Skip to content

typerdrive Cache modules

typerdrive.cache.attach

Provide a decorator that attaches the typerdrive cache to a typer command function.

Attributes

Classes

Functions

attach_cache 

attach_cache(
    show: bool = False,
) -> Callable[
    [ContextFunction[P, T]], ContextFunction[P, T]
]

Attach the typerdrive cache to the decorated typer command function.

Parameters:

Name Type Description Default
show bool

If set, show the cache after the function runs.

 False
Source code in src/typerdrive/cache/attach.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def attach_cache(show: bool = False) -> Callable[[ContextFunction[P, T]], ContextFunction[P, T]]:
    """
    Attach the `typerdrive` cache to the decorated `typer` command function.

    Parameters:
        show: If set, show the cache after the function runs.
    """

    def _decorate(func: ContextFunction[P, T]) -> ContextFunction[P, T]:
        manager_param_key: str | None = None
        for key in func.__annotations__.keys():
            if func.__annotations__[key] is CacheManager:
                func.__annotations__[key] = Annotated[CacheManager | None, CloakingDevice]
                manager_param_key = key

        @wraps(func)
        def wrapper(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs) -> T:
            manager: CacheManager = CacheManager()
            to_context(ctx, "cache_manager", manager)

            if manager_param_key:
                # ty: Cast kwargs to dict to allow mutation
                kwargs_dict = cast(dict[str, Any], kwargs)
                kwargs_dict[manager_param_key] = manager

            ret_val = func(ctx, *args, **kwargs)

            if show:
                manager.show(include_stats=False)

            return ret_val

        return wrapper

    return _decorate

get_cache_manager 

get_cache_manager(ctx: Context) -> CacheManager

Retrieve the CacheManager from the TyperdriveContext.

Source code in src/typerdrive/cache/attach.py 
17
18
19
20
21
22
23
24
25
26
27
def get_cache_manager(ctx: typer.Context) -> CacheManager:
    """
    Retrieve the `CacheManager` from the `TyperdriveContext`.
    """
    with CacheError.handle_errors("Cache is not bound to the context. Use the @attach_cache() decorator"):
        mgr: Any = from_context(ctx, "cache_manager")
    return CacheError.ensure_type(
        mgr,
        CacheManager,
        "Item in user context at `cache_manager` was not a CacheManager",
    )

typerdrive.cache.commands

Provide commands that can be added to a typer app to interact with the cache.

Classes

Functions

add_cache_subcommand 

add_cache_subcommand(cli: Typer)

Add all cache subcommands to the given app.

Source code in src/typerdrive/cache/commands.py 
80
81
82
83
84
85
86
87
88
89
def add_cache_subcommand(cli: typer.Typer):
    """
    Add all `cache` subcommands to the given app.
    """
    cache_cli = typer.Typer(help="Manage cache for the app")

    for cmd in [add_clear, add_show]:
        cmd(cache_cli)

    cli.add_typer(cache_cli, name="cache")

add_clear 

add_clear(cli: Typer)

Add the clear command to the given typer app.

Source code in src/typerdrive/cache/commands.py 
42
43
44
45
46
def add_clear(cli: typer.Typer):
    """
    Add the `clear` command to the given `typer` app.
    """
    cli.command()(clear)

add_show 

add_show(cli: Typer)

Add the show command to the given typer app.

Source code in src/typerdrive/cache/commands.py 
73
74
75
76
77
def add_show(cli: typer.Typer):
    """
    Add the `show` command to the given `typer` app.
    """
    cli.command()(show)

clear 

clear(
    ctx: Context,
    group: Annotated[
        str | None,
        Option(
            help="Clear only entries with this group. If not provided, clear entire cache"
        ),
    ] = None,
)

Remove multiple entries from the cache.

Parameters:

Name Type Description Default
group Annotated[str | None, Option(help='Clear only entries with this group. If not provided, clear entire cache')]

If provided, only remove entries with this group. Otherwise, clear entire cache.

 None
Source code in src/typerdrive/cache/commands.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@handle_errors("Failed to clear cache", handle_exc_class=CacheError)
@attach_cache()
def clear(
    ctx: typer.Context,
    group: Annotated[
        str | None,
        typer.Option(help="Clear only entries with this group. If not provided, clear entire cache"),
    ] = None,
):
    """
    Remove multiple entries from the cache.

    Parameters:
        group: If provided, only remove entries with this group. Otherwise, clear entire cache.
    """
    manager: CacheManager = get_cache_manager(ctx)

    if group:
        count = manager.clear(group=group)
        terminal_message(f"Cleared {count} entries with group '{group}'")
    else:
        typer.confirm("Are you sure you want to clear the entire cache?", abort=True)
        count = manager.clear()
        terminal_message(f"Cleared {count} entries from cache")

show 

show(
    ctx: Context,
    group: Annotated[
        str | None, Option(help="Filter entries by group")
    ] = None,
    stats: Annotated[
        bool,
        Option(
            help="Show cache statistics instead of entries"
        ),
    ] = False,
)

Show cache contents or statistics.

Parameters:

Name Type Description Default
group Annotated[str | None, Option(help='Filter entries by group')]

Optional group to filter entries

 None
stats Annotated[bool, Option(help='Show cache statistics instead of entries')]

If True, show statistics instead of entries

 False
Source code in src/typerdrive/cache/commands.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@handle_errors("Failed to show cache", handle_exc_class=CacheError)
@attach_cache()
def show(
    ctx: typer.Context,
    group: Annotated[
        str | None,
        typer.Option(help="Filter entries by group"),
    ] = None,
    stats: Annotated[
        bool,
        typer.Option(help="Show cache statistics instead of entries"),
    ] = False,
):
    """
    Show cache contents or statistics.

    Parameters:
        group: Optional group to filter entries
        stats: If True, show statistics instead of entries
    """
    manager: CacheManager = get_cache_manager(ctx)
    manager.show(group=group, show_stats=stats)

typerdrive.cache.exceptions

Provide exceptions specific to the cache feature of typerdrive.

Classes

CacheClearError

Bases: CacheError

Indicate that there was a problem clearing data from the cache.

Source code in src/typerdrive/cache/exceptions.py 
28
29
30
31
class CacheClearError(CacheError):
    """
    Indicate that there was a problem clearing data from the cache.
    """

CacheError

Bases: TyperdriveError

The base exception for cache errors.

Source code in src/typerdrive/cache/exceptions.py 
 9
10
11
12
13
class CacheError(TyperdriveError):
    """
    The base exception for cache errors.
    """
    exit_code: ExitCode = ExitCode.GENERAL_ERROR
Attributes
exit_code class-attribute instance-attribute 
exit_code: ExitCode = GENERAL_ERROR

CacheInitError

Bases: CacheError

Indicate that there was a problem initializing the cache.

Source code in src/typerdrive/cache/exceptions.py 
16
17
18
19
class CacheInitError(CacheError):
    """
    Indicate that there was a problem initializing the cache.
    """

CacheLoadError

Bases: CacheError

Indicate that there was a problem loading data from the cache.

Source code in src/typerdrive/cache/exceptions.py 
34
35
36
37
class CacheLoadError(CacheError):
    """
    Indicate that there was a problem loading data from the cache.
    """

CacheStoreError

Bases: CacheError

Indicate that there was a problem storing data in the cache.

Source code in src/typerdrive/cache/exceptions.py 
22
23
24
25
class CacheStoreError(CacheError):
    """
    Indicate that there was a problem storing data in the cache.
    """

typerdrive.cache.manager

Provide a class for managing the typerdrive cache feature using diskcache.

Classes

CacheManager

Manage the typerdrive cache feature using diskcache library.

This provides a traditional key-value cache with TTL support, eviction policies, and efficient disk-based storage.

Source code in src/typerdrive/cache/manager.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
296
297
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
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class CacheManager:
    """
    Manage the `typerdrive` cache feature using diskcache library.

    This provides a traditional key-value cache with TTL support, eviction policies,
    and efficient disk-based storage.
    """

    cache_dir: Path
    """ The directory where the cache database is stored. """

    cache: TypedCache
    """ The underlying TypedCache instance (wrapper around diskcache.Cache). """

    def __init__(
        self,
        size_limit: int = 2**30,
        eviction_policy: EvictionPolicy = EvictionPolicy.LEAST_RECENTLY_USED,
    ):
        """
        Initialize the cache manager.

        Parameters:
            size_limit: Maximum size of the cache in bytes (default: 1GB)
            eviction_policy: Eviction policy to use when size limit is reached
        """
        config: TyperdriveConfig = get_typerdrive_config()
        self.cache_dir = config.cache_dir

        with CacheInitError.handle_errors("Failed to initialize cache"):
            self.cache_dir.mkdir(parents=True, exist_ok=True)
            self.cache = TypedCache(
                directory=str(self.cache_dir),
                size_limit=size_limit,
                eviction_policy=eviction_policy.value,
            )

    def set(
        self,
        key: str,
        value: Any,
        expire: timedelta | None = None,
        group: str | None = None,
    ) -> bool:
        """
        Set a value in the cache.

        Parameters:
            key: The cache key
            value: The value to store (must be picklable)
            expire: Time until the key expires (None for no expiration)
            group: Optional group for organizing related cache entries

        Returns:
            True if the value was successfully set
        """
        logger.debug(f"Setting cache key: {key}")

        with CacheStoreError.handle_errors(f"Failed to store value for key '{key}'"):
            # Convert timedelta to seconds for diskcache
            expire_seconds = expire.total_seconds() if expire else None
            return self.cache.set(key, value, expire=expire_seconds, tag=group)

    def get(self, key: str, default: Any = None) -> Any:
        """
        Get a value from the cache.

        Parameters:
            key: The cache key
            default: Value to return if key is not found

        Returns:
            The cached value or default if not found
        """
        logger.debug(f"Getting cache key: {key}")

        with CacheLoadError.handle_errors(f"Failed to load value for key '{key}'"):
            return self.cache.get(key, default=default)

    def setdefault(
        self,
        key: str,
        default: Any = None,
        expire: timedelta | None = None,
        group: str | None = None,
    ) -> Any:
        """
        Get a value from the cache, setting it to default if not found.

        This mimics dict.setdefault() behavior: if the key exists in the cache,
        return its value. Otherwise, set the key to the default value and return it.

        Parameters:
            key: The cache key
            default: Value to set and return if key is not found
            expire: Time until the key expires (None for no expiration)
            group: Optional group for organizing related cache entries

        Returns:
            The cached value if it exists, otherwise the default value (which is also stored)
        """
        logger.debug(f"Getting or setting default for cache key: {key}")

        with CacheLoadError.handle_errors(f"Failed to get or set default for key '{key}'"):
            # Use a sentinel value to detect cache misses
            sentinel = object()
            value = self.cache.get(key, default=sentinel)

            if value is not sentinel:
                return value

            # Key doesn't exist, set the default value
            expire_seconds = expire.total_seconds() if expire else None
            self.cache.set(key, default, expire=expire_seconds, tag=group)
            return default

    def delete(self, key: str) -> bool:
        """
        Delete a key from the cache.

        Parameters:
            key: The cache key to delete

        Returns:
            True if the key was found and deleted
        """
        logger.debug(f"Deleting cache key: {key}")

        with CacheClearError.handle_errors(f"Failed to delete key '{key}'"):
            return self.cache.delete(key)

    def clear(self, group: str | None = None) -> int:
        """
        Remove items from the cache.

        Parameters:
            group: If provided, only remove entries with this group. Otherwise, remove all items.

        Returns:
            The number of items removed
        """
        if group:
            logger.debug(f"Evicting cache entries with group: {group}")
            with CacheClearError.handle_errors(f"Failed to evict group '{group}'"):
                return self.cache.evict(group)
        else:
            logger.debug("Clearing entire cache")
            with CacheClearError.handle_errors("Failed to clear cache"):
                count = len(self.cache)
                self.cache.clear()
                return count

    def keys(self, pattern: str | None = None, group: str | None = None) -> list[str]:
        """
        Get keys in the cache, optionally filtered by pattern and/or group.

        Parameters:
            pattern: Optional regex pattern to filter keys
            group: Optional group to filter by

        Returns:
            List of matching cache keys
        """
        with CacheError.handle_errors("Failed to retrieve cache keys"):
            # Convert keys to strings (diskcache may return bytes)
            all_keys: list[str] = [k.decode("utf-8") if isinstance(k, bytes) else k for k in self.cache.iterkeys()]

            # Filter by pattern if provided
            if pattern:
                regex = re.compile(pattern)
                all_keys = [k for k in all_keys if regex.search(k)]

            # Filter by group if provided
            # Note: diskcache doesn't provide a way to query keys by group directly,
            # so we need to check each key's group
            if group:
                filtered_keys: list[str] = []
                for key in all_keys:
                    try:
                        # TypedCache.get() with tag=True returns tuple[value, tag_str]
                        _, key_group = self.cache.get(key, tag=True)
                        if key_group == group:
                            filtered_keys.append(key)
                    except Exception:
                        pass
                all_keys = filtered_keys

            return all_keys

    def get_group(self, key: str) -> str | None:
        """
        Get the group associated with a key.

        Parameters:
            key: The cache key

        Returns:
            The group associated with the key, or None if no group or key doesn't exist
        """
        with CacheError.handle_errors(f"Failed to get group for key '{key}'"):
            # Use TypedCache's tag parameter to retrieve the group
            # When tag=True, get() returns (value, tag) tuple
            try:
                _, group = self.cache.get(key, tag=True)
                return group
            except KeyError:
                return None

    def get_ttl(self, key: str) -> str:
        """
        Get the time-to-live for a cache key in human-readable format.

        Parameters:
            key: The cache key

        Returns:
            Human-readable TTL string (e.g., "2 hours", "never"), or "expired" if the key doesn't exist
        """
        with CacheError.handle_errors(f"Failed to get TTL for key '{key}'"):
            # Get expire_time from TypedCache
            # TypedCache.get() with expire_time=True returns tuple[value, expire_time_float]
            value, expire_time = self.cache.get(key, default=None, expire_time=True)

            # If value is None, the key doesn't exist
            if value is None:
                return "expired"

            # expire_time is None if no expiration set
            if expire_time is None:
                return "never"

            # Calculate remaining time
            current_time = time.time()
            remaining_seconds = expire_time - current_time

            if remaining_seconds <= 0:
                return "expired"

            # Use humanize for human-readable format
            return humanize.naturaldelta(timedelta(seconds=remaining_seconds))

    def stats(self) -> CacheStats:
        """
        Get cache statistics.

        Returns:
            CacheStats object with cache statistics
        """
        with CacheError.handle_errors("Failed to retrieve cache stats"):
            # TypedCache.stats() returns tuple[int, int] for (hits, misses)
            hits, misses = self.cache.stats(enable=True)
            return CacheStats(
                hits=hits,
                misses=misses,
                size=len(self.cache),
                volume=self.cache.volume(),
            )

    def show(
        self,
        pattern: str | None = None,
        group: str | None = None,
        show_stats: bool = False,
        include_stats: bool = True,
    ) -> None:
        """
        Display cache contents or statistics.

        Parameters:
            pattern: Optional regex pattern to filter keys
            group: Optional group to filter entries
            show_stats: If True, show statistics instead of entries
            include_stats: If True and show_stats is False, include stats summary at bottom of entries
        """
        from rich.console import Group as RichGroup

        if show_stats:
            stats_data = self.stats()
            table = Table(show_header=True, header_style="bold", box=box.SIMPLE)
            table.add_column("Metric")
            table.add_column("Value")

            table.add_row("Size (entries)", str(stats_data.size))
            table.add_row("Volume (bytes)", str(stats_data.volume))
            table.add_row("Hits", str(stats_data.hits))
            table.add_row("Misses", str(stats_data.misses))

            terminal_message(table, subject="Cache Statistics")
        else:
            keys = self.keys(pattern=pattern, group=group)
            if not keys:
                terminal_message("No cache entries found")
                return

            # Build filter string for title
            filters: list[str] = []
            if pattern:
                filters.append(f"pattern={pattern}")
            if group:
                filters.append(f"group={group}")
            filter_str = f" ({', '.join(filters)})" if filters else ""

            # Build entries table
            entries_table = Table(show_header=True, header_style="bold", box=box.SIMPLE)
            entries_table.add_column("Key", style="green")
            entries_table.add_column("Group", style="yellow")
            entries_table.add_column("TTL", style="blue")

            for key in sorted(keys):
                key_group = self.get_group(key) or ""
                key_ttl = self.get_ttl(key)
                entries_table.add_row(key, key_group, key_ttl)

            if include_stats:
                # Get stats
                stats_data = self.stats()

                # Build stats table
                stats_table = Table(show_header=True, header_style="bold", box=box.SIMPLE)
                stats_table.add_column("Entries", justify="right")
                stats_table.add_column("Volume", justify="right")
                stats_table.add_column("Hits", justify="right")
                stats_table.add_column("Misses", justify="right")
                stats_table.add_row(
                    str(stats_data.size),
                    f"{stats_data.volume:,} bytes",
                    str(stats_data.hits),
                    str(stats_data.misses),
                )

                # Combine tables
                combined = RichGroup(entries_table, "", stats_table)
                terminal_message(combined, subject=f"Cache contains {len(keys)} entries{filter_str}")
            else:
                # Just show entries table
                terminal_message(entries_table, subject=f"Cache contains {len(keys)} entries{filter_str}")
Attributes
cache instance-attribute 
cache: TypedCache = TypedCache(
    directory=str(cache_dir),
    size_limit=size_limit,
    eviction_policy=value,
)

The underlying TypedCache instance (wrapper around diskcache.Cache).

cache_dir instance-attribute 
cache_dir: Path = cache_dir

The directory where the cache database is stored.

Functions
__init__ 
__init__(
    size_limit: int = 2**30,
    eviction_policy: EvictionPolicy = EvictionPolicy.LEAST_RECENTLY_USED,
)

Initialize the cache manager.

Parameters:

Name Type Description Default
size_limit int

Maximum size of the cache in bytes (default: 1GB)

 2 ** 30
eviction_policy EvictionPolicy

Eviction policy to use when size limit is reached

 LEAST_RECENTLY_USED
Source code in src/typerdrive/cache/manager.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    size_limit: int = 2**30,
    eviction_policy: EvictionPolicy = EvictionPolicy.LEAST_RECENTLY_USED,
):
    """
    Initialize the cache manager.

    Parameters:
        size_limit: Maximum size of the cache in bytes (default: 1GB)
        eviction_policy: Eviction policy to use when size limit is reached
    """
    config: TyperdriveConfig = get_typerdrive_config()
    self.cache_dir = config.cache_dir

    with CacheInitError.handle_errors("Failed to initialize cache"):
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.cache = TypedCache(
            directory=str(self.cache_dir),
            size_limit=size_limit,
            eviction_policy=eviction_policy.value,
        )
clear 
clear(group: str | None = None) -> int

Remove items from the cache.

Parameters:

Name Type Description Default
group str | None

If provided, only remove entries with this group. Otherwise, remove all items.

 None

Returns:

Type Description
int

The number of items removed
Source code in src/typerdrive/cache/manager.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def clear(self, group: str | None = None) -> int:
    """
    Remove items from the cache.

    Parameters:
        group: If provided, only remove entries with this group. Otherwise, remove all items.

    Returns:
        The number of items removed
    """
    if group:
        logger.debug(f"Evicting cache entries with group: {group}")
        with CacheClearError.handle_errors(f"Failed to evict group '{group}'"):
            return self.cache.evict(group)
    else:
        logger.debug("Clearing entire cache")
        with CacheClearError.handle_errors("Failed to clear cache"):
            count = len(self.cache)
            self.cache.clear()
            return count
delete 
delete(key: str) -> bool

Delete a key from the cache.

Parameters:

Name Type Description Default
key str

The cache key to delete

 required

Returns:

Type Description
bool

True if the key was found and deleted
Source code in src/typerdrive/cache/manager.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def delete(self, key: str) -> bool:
    """
    Delete a key from the cache.

    Parameters:
        key: The cache key to delete

    Returns:
        True if the key was found and deleted
    """
    logger.debug(f"Deleting cache key: {key}")

    with CacheClearError.handle_errors(f"Failed to delete key '{key}'"):
        return self.cache.delete(key)
get 
get(key: str, default: Any = None) -> Any

Get a value from the cache.

Parameters:

Name Type Description Default
key str

The cache key

 required
default Any

Value to return if key is not found

 None

Returns:

Type Description
Any

The cached value or default if not found
Source code in src/typerdrive/cache/manager.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get(self, key: str, default: Any = None) -> Any:
    """
    Get a value from the cache.

    Parameters:
        key: The cache key
        default: Value to return if key is not found

    Returns:
        The cached value or default if not found
    """
    logger.debug(f"Getting cache key: {key}")

    with CacheLoadError.handle_errors(f"Failed to load value for key '{key}'"):
        return self.cache.get(key, default=default)
get_group 
get_group(key: str) -> str | None

Get the group associated with a key.

Parameters:

Name Type Description Default
key str

The cache key

 required

Returns:

Type Description
str | None

The group associated with the key, or None if no group or key doesn't exist
Source code in src/typerdrive/cache/manager.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def get_group(self, key: str) -> str | None:
    """
    Get the group associated with a key.

    Parameters:
        key: The cache key

    Returns:
        The group associated with the key, or None if no group or key doesn't exist
    """
    with CacheError.handle_errors(f"Failed to get group for key '{key}'"):
        # Use TypedCache's tag parameter to retrieve the group
        # When tag=True, get() returns (value, tag) tuple
        try:
            _, group = self.cache.get(key, tag=True)
            return group
        except KeyError:
            return None
get_ttl 
get_ttl(key: str) -> str

Get the time-to-live for a cache key in human-readable format.

Parameters:

Name Type Description Default
key str

The cache key

 required

Returns:

Type Description
str

Human-readable TTL string (e.g., "2 hours", "never"), or "expired" if the key doesn't exist
Source code in src/typerdrive/cache/manager.py 
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
def get_ttl(self, key: str) -> str:
    """
    Get the time-to-live for a cache key in human-readable format.

    Parameters:
        key: The cache key

    Returns:
        Human-readable TTL string (e.g., "2 hours", "never"), or "expired" if the key doesn't exist
    """
    with CacheError.handle_errors(f"Failed to get TTL for key '{key}'"):
        # Get expire_time from TypedCache
        # TypedCache.get() with expire_time=True returns tuple[value, expire_time_float]
        value, expire_time = self.cache.get(key, default=None, expire_time=True)

        # If value is None, the key doesn't exist
        if value is None:
            return "expired"

        # expire_time is None if no expiration set
        if expire_time is None:
            return "never"

        # Calculate remaining time
        current_time = time.time()
        remaining_seconds = expire_time - current_time

        if remaining_seconds <= 0:
            return "expired"

        # Use humanize for human-readable format
        return humanize.naturaldelta(timedelta(seconds=remaining_seconds))
keys 
keys(
    pattern: str | None = None, group: str | None = None
) -> list[str]

Get keys in the cache, optionally filtered by pattern and/or group.

Parameters:

Name Type Description Default
pattern str | None

Optional regex pattern to filter keys

 None
group str | None

Optional group to filter by

 None

Returns:

Type Description
list[str]

List of matching cache keys
Source code in src/typerdrive/cache/manager.py 
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
def keys(self, pattern: str | None = None, group: str | None = None) -> list[str]:
    """
    Get keys in the cache, optionally filtered by pattern and/or group.

    Parameters:
        pattern: Optional regex pattern to filter keys
        group: Optional group to filter by

    Returns:
        List of matching cache keys
    """
    with CacheError.handle_errors("Failed to retrieve cache keys"):
        # Convert keys to strings (diskcache may return bytes)
        all_keys: list[str] = [k.decode("utf-8") if isinstance(k, bytes) else k for k in self.cache.iterkeys()]

        # Filter by pattern if provided
        if pattern:
            regex = re.compile(pattern)
            all_keys = [k for k in all_keys if regex.search(k)]

        # Filter by group if provided
        # Note: diskcache doesn't provide a way to query keys by group directly,
        # so we need to check each key's group
        if group:
            filtered_keys: list[str] = []
            for key in all_keys:
                try:
                    # TypedCache.get() with tag=True returns tuple[value, tag_str]
                    _, key_group = self.cache.get(key, tag=True)
                    if key_group == group:
                        filtered_keys.append(key)
                except Exception:
                    pass
            all_keys = filtered_keys

        return all_keys
set 
set(
    key: str,
    value: Any,
    expire: timedelta | None = None,
    group: str | None = None,
) -> bool

Set a value in the cache.

Parameters:

Name Type Description Default
key str

The cache key

 required
value Any

The value to store (must be picklable)

 required
expire timedelta | None

Time until the key expires (None for no expiration)

 None
group str | None

Optional group for organizing related cache entries

 None

Returns:

Type Description
bool

True if the value was successfully set
Source code in src/typerdrive/cache/manager.py 
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
def set(
    self,
    key: str,
    value: Any,
    expire: timedelta | None = None,
    group: str | None = None,
) -> bool:
    """
    Set a value in the cache.

    Parameters:
        key: The cache key
        value: The value to store (must be picklable)
        expire: Time until the key expires (None for no expiration)
        group: Optional group for organizing related cache entries

    Returns:
        True if the value was successfully set
    """
    logger.debug(f"Setting cache key: {key}")

    with CacheStoreError.handle_errors(f"Failed to store value for key '{key}'"):
        # Convert timedelta to seconds for diskcache
        expire_seconds = expire.total_seconds() if expire else None
        return self.cache.set(key, value, expire=expire_seconds, tag=group)
setdefault 
setdefault(
    key: str,
    default: Any = None,
    expire: timedelta | None = None,
    group: str | None = None,
) -> Any

Get a value from the cache, setting it to default if not found.

This mimics dict.setdefault() behavior: if the key exists in the cache, return its value. Otherwise, set the key to the default value and return it.

Parameters:

Name Type Description Default
key str

The cache key

 required
default Any

Value to set and return if key is not found

 None
expire timedelta | None

Time until the key expires (None for no expiration)

 None
group str | None

Optional group for organizing related cache entries

 None

Returns:

Type Description
Any

The cached value if it exists, otherwise the default value (which is also stored)
Source code in src/typerdrive/cache/manager.py 
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
def setdefault(
    self,
    key: str,
    default: Any = None,
    expire: timedelta | None = None,
    group: str | None = None,
) -> Any:
    """
    Get a value from the cache, setting it to default if not found.

    This mimics dict.setdefault() behavior: if the key exists in the cache,
    return its value. Otherwise, set the key to the default value and return it.

    Parameters:
        key: The cache key
        default: Value to set and return if key is not found
        expire: Time until the key expires (None for no expiration)
        group: Optional group for organizing related cache entries

    Returns:
        The cached value if it exists, otherwise the default value (which is also stored)
    """
    logger.debug(f"Getting or setting default for cache key: {key}")

    with CacheLoadError.handle_errors(f"Failed to get or set default for key '{key}'"):
        # Use a sentinel value to detect cache misses
        sentinel = object()
        value = self.cache.get(key, default=sentinel)

        if value is not sentinel:
            return value

        # Key doesn't exist, set the default value
        expire_seconds = expire.total_seconds() if expire else None
        self.cache.set(key, default, expire=expire_seconds, tag=group)
        return default
show 
show(
    pattern: str | None = None,
    group: str | None = None,
    show_stats: bool = False,
    include_stats: bool = True,
) -> None

Display cache contents or statistics.

Parameters:

Name Type Description Default
pattern str | None

Optional regex pattern to filter keys

 None
group str | None

Optional group to filter entries

 None
show_stats bool

If True, show statistics instead of entries

 False
include_stats bool

If True and show_stats is False, include stats summary at bottom of entries

 True
Source code in src/typerdrive/cache/manager.py 
289
290
291
292
293
294
295
296
297
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
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def show(
    self,
    pattern: str | None = None,
    group: str | None = None,
    show_stats: bool = False,
    include_stats: bool = True,
) -> None:
    """
    Display cache contents or statistics.

    Parameters:
        pattern: Optional regex pattern to filter keys
        group: Optional group to filter entries
        show_stats: If True, show statistics instead of entries
        include_stats: If True and show_stats is False, include stats summary at bottom of entries
    """
    from rich.console import Group as RichGroup

    if show_stats:
        stats_data = self.stats()
        table = Table(show_header=True, header_style="bold", box=box.SIMPLE)
        table.add_column("Metric")
        table.add_column("Value")

        table.add_row("Size (entries)", str(stats_data.size))
        table.add_row("Volume (bytes)", str(stats_data.volume))
        table.add_row("Hits", str(stats_data.hits))
        table.add_row("Misses", str(stats_data.misses))

        terminal_message(table, subject="Cache Statistics")
    else:
        keys = self.keys(pattern=pattern, group=group)
        if not keys:
            terminal_message("No cache entries found")
            return

        # Build filter string for title
        filters: list[str] = []
        if pattern:
            filters.append(f"pattern={pattern}")
        if group:
            filters.append(f"group={group}")
        filter_str = f" ({', '.join(filters)})" if filters else ""

        # Build entries table
        entries_table = Table(show_header=True, header_style="bold", box=box.SIMPLE)
        entries_table.add_column("Key", style="green")
        entries_table.add_column("Group", style="yellow")
        entries_table.add_column("TTL", style="blue")

        for key in sorted(keys):
            key_group = self.get_group(key) or ""
            key_ttl = self.get_ttl(key)
            entries_table.add_row(key, key_group, key_ttl)

        if include_stats:
            # Get stats
            stats_data = self.stats()

            # Build stats table
            stats_table = Table(show_header=True, header_style="bold", box=box.SIMPLE)
            stats_table.add_column("Entries", justify="right")
            stats_table.add_column("Volume", justify="right")
            stats_table.add_column("Hits", justify="right")
            stats_table.add_column("Misses", justify="right")
            stats_table.add_row(
                str(stats_data.size),
                f"{stats_data.volume:,} bytes",
                str(stats_data.hits),
                str(stats_data.misses),
            )

            # Combine tables
            combined = RichGroup(entries_table, "", stats_table)
            terminal_message(combined, subject=f"Cache contains {len(keys)} entries{filter_str}")
        else:
            # Just show entries table
            terminal_message(entries_table, subject=f"Cache contains {len(keys)} entries{filter_str}")
stats 
stats() -> CacheStats

Get cache statistics.

Returns:

Type Description
CacheStats

CacheStats object with cache statistics
Source code in src/typerdrive/cache/manager.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def stats(self) -> CacheStats:
    """
    Get cache statistics.

    Returns:
        CacheStats object with cache statistics
    """
    with CacheError.handle_errors("Failed to retrieve cache stats"):
        # TypedCache.stats() returns tuple[int, int] for (hits, misses)
        hits, misses = self.cache.stats(enable=True)
        return CacheStats(
            hits=hits,
            misses=misses,
            size=len(self.cache),
            volume=self.cache.volume(),
        )

Functions