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 
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
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:
                kwargs[manager_param_key] = manager

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

            if show:
                show_directory(manager.cache_dir, subject="Current cache")

            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 
18
19
20
21
22
23
24
25
26
27
28
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 
64
65
66
67
68
69
70
71
72
73
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 
41
42
43
44
45
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 
57
58
59
60
61
def add_show(cli: typer.Typer):
    """
    Add the `show` command to the given `typer` app.
    """
    cli.command()(show)

clear 

clear(
    ctx: Context,
    path: Annotated[
        str | None,
        Option(
            help="Clear only the entry matching this path. If not provided, clear the entire cache"
        ),
    ] = None,
)

Remove data from the cache.

Parameters:

Name Type Description Default
path Annotated[str | None, Option(help='Clear only the entry matching this path. If not provided, clear the entire cache')]

If provided, only remove the data at the given cache key. Otherwise, clear the 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
@handle_errors("Failed to clear cache", handle_exc_class=CacheError)
@attach_cache()
def clear(
    ctx: typer.Context,
    path: Annotated[
        str | None,
        typer.Option(help="Clear only the entry matching this path. If not provided, clear the entire cache"),
    ] = None,
):
    """
    Remove data from the cache.

    Parameters:
        path: If provided, only remove the data at the given cache key. Otherwise, clear the entire cache.
    """
    manager: CacheManager = get_cache_manager(ctx)
    if path:
        manager.clear_path(path)
        terminal_message(f"Cleared entry at cache target {str(path)}")
    else:
        typer.confirm("Are you sure you want to clear the entire cache?", abort=True)
        count = manager.clear_all()
        terminal_message(f"Cleared all {count} files from cache")

show 

show(ctx: Context)

Show the current cache directory.

Source code in src/typerdrive/cache/commands.py 
48
49
50
51
52
53
54
@handle_errors("Failed to show cache", handle_exc_class=CacheError)
@attach_cache(show=True)
def show(ctx: typer.Context):  # pyright: ignore[reportUnusedParameter]
    """
    Show the current cache directory.
    """
    pass

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.

Classes

CacheManager

Manage the typerdrive cache feature.

Source code in src/typerdrive/cache/manager.py 
 25
 26
 27
 28
 29
 30
 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
class CacheManager:
    """
    Manage the `typerdrive` cache feature.
    """

    cache_dir: Path
    """ The directory where the cache is found. """

    def __init__(self):
        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)

    def resolve_path(self, path: Path | str, mkdir: bool = False) -> Path:
        """
        Resolve a given cache key path to an absolute path within the cache directory.

        If the resolved path is outside the cache directory, an exception will be raised.
        If the resolved path is the same as the cache directory, an exception will be raised.

        Parameters:
            path:  The cache key
            mkdir: If set, create the directory for the cache key if it does not exist.
        """
        if isinstance(path, str):
            path = Path(path)
        full_path = self.cache_dir / path
        full_path = full_path.resolve()
        CacheError.require_condition(
            is_child(full_path, self.cache_dir),
            f"Resolved cache path {str(full_path)} is not within cache {str(self.cache_dir)}",
        )
        CacheError.require_condition(
            full_path != self.cache_dir,
            f"Resolved cache path {str(full_path)} must not be the same as cache {str(self.cache_dir)}",
        )
        if mkdir:
            full_path.parent.mkdir(parents=True, exist_ok=True)
        return full_path

    def list_items(self, path: Path | str) -> list[str]:
        """
        List items at a given path.

        Items will be all non-directory entries in the given path.

        If the path doesn't exist or is not a directory, an exception will be raised.
        """
        full_path = self.resolve_path(path)
        CacheError.require_condition(
            full_path.exists(),
            f"Resolved cache path {str(full_path)} does not exist",
        )
        CacheError.require_condition(
            full_path.is_dir(),
            "Resolved cache path {str(full_path)} is not a directory",
        )
        items: list[str] = []
        for path in full_path.iterdir():
            if path.is_dir():
                continue
            items.append(str(path.name))
        return items

    def store_bytes(self, data: bytes, path: Path | str, mode: int | None = None):
        """
        Store data at the given cache key.

        Parameters:
            data: The data to store in the cache
            path: The cache key where the data should be stored
            mode: The file mode to use when creating the cache entry
        """
        full_path = self.resolve_path(path, mkdir=True)

        logger.debug(f"Storing data at {full_path}")

        with CacheStoreError.handle_errors(f"Failed to store data in cache target {str(path)}"):
            full_path.write_bytes(data)
        if mode:
            with CacheStoreError.handle_errors(f"Failed to set mode for cache target {str(path)} to {mode=}"):
                full_path.chmod(mode)

    def store_text(self, text: str, path: Path | str, mode: int | None = None):
        """
        Store text at the given cache key.

        Parameters:
            text: The text to store in the cache
            path: The cache key where the text should be stored
            mode: The file mode to use when creating the cache entry
        """
        self.store_bytes(text.encode("utf-8"), path, mode=mode)

    def store_json(self, data: dict[str, Any], path: Path | str, mode: int | None = None):
        """
        Store a dictionary at the given cache key.

        The dictionary must be json serializable.

        Parameters:
            data: The dict to store in the cache
            path: The cache key where the dict should be stored
            mode: The file mode to use when creating the cache entry
        """
        self.store_bytes(json.dumps(data, indent=2).encode("utf-8"), path, mode=mode)

    def load_bytes(self, path: Path | str) -> bytes:
        """
        Load data from the cache at the given key.

        If there is no data at the given key, an exception will be raised.
        If there is an error reading the data, an exception will be raised.

        Parameters:
            path: The cache key where the data should be loaded from
        """
        full_path = self.resolve_path(path, mkdir=False)

        logger.debug(f"Loading data from {full_path}")

        CacheLoadError.require_condition(full_path.exists(), f"Cache target {str(path)} does not exist")
        with CacheLoadError.handle_errors(f"Failed to load data from cache target {str(path)}"):
            return full_path.read_bytes()

    def load_text(self, path: Path | str) -> str:
        """
        Load text from the cache at the given key.

        Parameters:
            path: The cache key where the text should be loaded from
        """
        return self.load_bytes(path).decode("utf-8")

    def load_json(self, path: Path | str) -> dict[str, Any]:
        """
        Load a dictionary from the cache at the given key.

        The data at the given key must be json deserializable.

        Parameters:
            path: The cache key where the dict should be loaded from
        """
        text = self.load_bytes(path).decode("utf-8")
        with CacheLoadError.handle_errors(f"Failed to unpack JSON data from cache target {str(path)}"):
            return json.loads(text)

    def clear_path(self, path: Path | str) -> Path:
        """
        Removes data from the cache at the given key.

        Parameters:
            path: The cache key where the data should be cleared
        """
        full_path = self.resolve_path(path)

        logger.debug(f"Clearing data at {full_path}")

        with CacheClearError.handle_errors(f"Failed to clear cache target {str(path)}"):
            full_path.unlink()
        if len([p for p in full_path.parent.iterdir()]) == 0:
            with CacheClearError.handle_errors(f"Failed to remove empty directory {str(full_path.parent)}"):
                full_path.parent.rmdir()
        return full_path

    def clear_all(self) -> int:
        """
        Removes all data from the cache.
        """
        logger.debug("Clearing entire cache")
        with CacheClearError.handle_errors("Failed to clear cache"):
            return clear_directory(self.cache_dir)
Attributes
cache_dir instance-attribute 
cache_dir: Path = cache_dir

The directory where the cache is found.

Functions
__init__ 
__init__()
Source code in src/typerdrive/cache/manager.py 
33
34
35
36
37
38
39
def __init__(self):
    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)
clear_all 
clear_all() -> int

Removes all data from the cache.

Source code in src/typerdrive/cache/manager.py 
193
194
195
196
197
198
199
def clear_all(self) -> int:
    """
    Removes all data from the cache.
    """
    logger.debug("Clearing entire cache")
    with CacheClearError.handle_errors("Failed to clear cache"):
        return clear_directory(self.cache_dir)
clear_path 
clear_path(path: Path | str) -> Path

Removes data from the cache at the given key.

Parameters:

Name Type Description Default
path Path | str

The cache key where the data should be cleared

 required
Source code in src/typerdrive/cache/manager.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def clear_path(self, path: Path | str) -> Path:
    """
    Removes data from the cache at the given key.

    Parameters:
        path: The cache key where the data should be cleared
    """
    full_path = self.resolve_path(path)

    logger.debug(f"Clearing data at {full_path}")

    with CacheClearError.handle_errors(f"Failed to clear cache target {str(path)}"):
        full_path.unlink()
    if len([p for p in full_path.parent.iterdir()]) == 0:
        with CacheClearError.handle_errors(f"Failed to remove empty directory {str(full_path.parent)}"):
            full_path.parent.rmdir()
    return full_path
list_items 
list_items(path: Path | str) -> list[str]

List items at a given path.

Items will be all non-directory entries in the given path.

If the path doesn't exist or is not a directory, an exception will be raised.

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
def list_items(self, path: Path | str) -> list[str]:
    """
    List items at a given path.

    Items will be all non-directory entries in the given path.

    If the path doesn't exist or is not a directory, an exception will be raised.
    """
    full_path = self.resolve_path(path)
    CacheError.require_condition(
        full_path.exists(),
        f"Resolved cache path {str(full_path)} does not exist",
    )
    CacheError.require_condition(
        full_path.is_dir(),
        "Resolved cache path {str(full_path)} is not a directory",
    )
    items: list[str] = []
    for path in full_path.iterdir():
        if path.is_dir():
            continue
        items.append(str(path.name))
    return items
load_bytes 
load_bytes(path: Path | str) -> bytes

Load data from the cache at the given key.

If there is no data at the given key, an exception will be raised. If there is an error reading the data, an exception will be raised.

Parameters:

Name Type Description Default
path Path | str

The cache key where the data should be loaded from

 required
Source code in src/typerdrive/cache/manager.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def load_bytes(self, path: Path | str) -> bytes:
    """
    Load data from the cache at the given key.

    If there is no data at the given key, an exception will be raised.
    If there is an error reading the data, an exception will be raised.

    Parameters:
        path: The cache key where the data should be loaded from
    """
    full_path = self.resolve_path(path, mkdir=False)

    logger.debug(f"Loading data from {full_path}")

    CacheLoadError.require_condition(full_path.exists(), f"Cache target {str(path)} does not exist")
    with CacheLoadError.handle_errors(f"Failed to load data from cache target {str(path)}"):
        return full_path.read_bytes()
load_json 
load_json(path: Path | str) -> dict[str, Any]

Load a dictionary from the cache at the given key.

The data at the given key must be json deserializable.

Parameters:

Name Type Description Default
path Path | str

The cache key where the dict should be loaded from

 required
Source code in src/typerdrive/cache/manager.py 
162
163
164
165
166
167
168
169
170
171
172
173
def load_json(self, path: Path | str) -> dict[str, Any]:
    """
    Load a dictionary from the cache at the given key.

    The data at the given key must be json deserializable.

    Parameters:
        path: The cache key where the dict should be loaded from
    """
    text = self.load_bytes(path).decode("utf-8")
    with CacheLoadError.handle_errors(f"Failed to unpack JSON data from cache target {str(path)}"):
        return json.loads(text)
load_text 
load_text(path: Path | str) -> str

Load text from the cache at the given key.

Parameters:

Name Type Description Default
path Path | str

The cache key where the text should be loaded from

 required
Source code in src/typerdrive/cache/manager.py 
153
154
155
156
157
158
159
160
def load_text(self, path: Path | str) -> str:
    """
    Load text from the cache at the given key.

    Parameters:
        path: The cache key where the text should be loaded from
    """
    return self.load_bytes(path).decode("utf-8")
resolve_path 
resolve_path(path: Path | str, mkdir: bool = False) -> Path

Resolve a given cache key path to an absolute path within the cache directory.

If the resolved path is outside the cache directory, an exception will be raised. If the resolved path is the same as the cache directory, an exception will be raised.

Parameters:

Name Type Description Default
path Path | str

The cache key

 required
mkdir bool

If set, create the directory for the cache key if it does not exist.

 False
Source code in src/typerdrive/cache/manager.py 
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
def resolve_path(self, path: Path | str, mkdir: bool = False) -> Path:
    """
    Resolve a given cache key path to an absolute path within the cache directory.

    If the resolved path is outside the cache directory, an exception will be raised.
    If the resolved path is the same as the cache directory, an exception will be raised.

    Parameters:
        path:  The cache key
        mkdir: If set, create the directory for the cache key if it does not exist.
    """
    if isinstance(path, str):
        path = Path(path)
    full_path = self.cache_dir / path
    full_path = full_path.resolve()
    CacheError.require_condition(
        is_child(full_path, self.cache_dir),
        f"Resolved cache path {str(full_path)} is not within cache {str(self.cache_dir)}",
    )
    CacheError.require_condition(
        full_path != self.cache_dir,
        f"Resolved cache path {str(full_path)} must not be the same as cache {str(self.cache_dir)}",
    )
    if mkdir:
        full_path.parent.mkdir(parents=True, exist_ok=True)
    return full_path
store_bytes 
store_bytes(
    data: bytes, path: Path | str, mode: int | None = None
)

Store data at the given cache key.

Parameters:

Name Type Description Default
data bytes

The data to store in the cache

 required
path Path | str

The cache key where the data should be stored

 required
mode int | None

The file mode to use when creating the cache entry

 None
Source code in src/typerdrive/cache/manager.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def store_bytes(self, data: bytes, path: Path | str, mode: int | None = None):
    """
    Store data at the given cache key.

    Parameters:
        data: The data to store in the cache
        path: The cache key where the data should be stored
        mode: The file mode to use when creating the cache entry
    """
    full_path = self.resolve_path(path, mkdir=True)

    logger.debug(f"Storing data at {full_path}")

    with CacheStoreError.handle_errors(f"Failed to store data in cache target {str(path)}"):
        full_path.write_bytes(data)
    if mode:
        with CacheStoreError.handle_errors(f"Failed to set mode for cache target {str(path)} to {mode=}"):
            full_path.chmod(mode)
store_json 
store_json(
    data: dict[str, Any],
    path: Path | str,
    mode: int | None = None,
)

Store a dictionary at the given cache key.

The dictionary must be json serializable.

Parameters:

Name Type Description Default
data dict[str, Any]

The dict to store in the cache

 required
path Path | str

The cache key where the dict should be stored

 required
mode int | None

The file mode to use when creating the cache entry

 None
Source code in src/typerdrive/cache/manager.py 
122
123
124
125
126
127
128
129
130
131
132
133
def store_json(self, data: dict[str, Any], path: Path | str, mode: int | None = None):
    """
    Store a dictionary at the given cache key.

    The dictionary must be json serializable.

    Parameters:
        data: The dict to store in the cache
        path: The cache key where the dict should be stored
        mode: The file mode to use when creating the cache entry
    """
    self.store_bytes(json.dumps(data, indent=2).encode("utf-8"), path, mode=mode)
store_text 
store_text(
    text: str, path: Path | str, mode: int | None = None
)

Store text at the given cache key.

Parameters:

Name Type Description Default
text str

The text to store in the cache

 required
path Path | str

The cache key where the text should be stored

 required
mode int | None

The file mode to use when creating the cache entry

 None
Source code in src/typerdrive/cache/manager.py 
111
112
113
114
115
116
117
118
119
120
def store_text(self, text: str, path: Path | str, mode: int | None = None):
    """
    Store text at the given cache key.

    Parameters:
        text: The text to store in the cache
        path: The cache key where the text should be stored
        mode: The file mode to use when creating the cache entry
    """
    self.store_bytes(text.encode("utf-8"), path, mode=mode)

Functions