checkpointing.cache.pickle_file

View Source

 1from checkpointing.cache.base import CacheBase
 2from checkpointing._typing import ReturnValue
 3from checkpointing import defaults
 4from checkpointing.exceptions import CheckpointNotExist
 5import pathlib
 6import os
 7from checkpointing.util import pickle
 8
 9
10class PickleFileCache(CacheBase):
11    """Cache the result as pickle files saved on the disk."""
12
13    def __init__(self, directory: os.PathLike = None, pickle_protocol: int = None) -> None:
14        """
15        Args:
16            directory: the directory where the files will be saved. The directory will be created if it does not exist.
17                       If None, use the global default `cache.filesystem.directory`
18            pickle_protocol: the protocol used when pickling files. If None, use the global default
19                             `cache.pickle_protocol`
20        """
21
22        self.__directory = pathlib.Path(directory if directory is not None else defaults["cache.filesystem.directory"])
23        self.__directory.mkdir(parents=True, exist_ok=True)
24
25        self.__pickle_protocol = pickle_protocol if pickle_protocol is not None else defaults["cache.pickle_protocol"]
26
27    def _get_file_path(self, context_id: str) -> pathlib.Path:
28        """
29        Args:
30            context_id: the file name without the file extension
31
32        Returns:
33            The full file path used in this cache that corresponds to the context id
34        """
35
36        filename = f"{context_id}.pickle"
37        return self.__directory.joinpath(filename)
38
39    def save(self, context_id: str, result: ReturnValue) -> None:
40        """
41        Save the result with the given context id.
42
43        Args:
44            context_id: identifier of the function call context, must be a valid file name without the file extension
45            result: return value of the function call
46        """
47
48        with open(self._get_file_path(context_id), mode="wb") as file:
49            pickle.dump(result, file, protocol=self.__pickle_protocol)
50
51    def retrieve(self, context_id: str) -> ReturnValue:
52        """
53        Retrieve the function return value with the given context id.
54        If there is no cached results for the context_id, throws a checkpointing.exceptions.CheckpointNotExist
55
56        Args:
57            context_id: identifier of the function call context, must be a valid file name without the file extension
58
59        Returns:
60            The return value of the function that corresponds to this context id
61        """
62
63        path = self._get_file_path(context_id)
64        if not path.exists():
65            raise CheckpointNotExist
66
67        with open(path, mode="rb") as file:
68            return pickle.load(file, protocol=self.__pickle_protocol)

class PickleFileCache(abc.ABC, typing.Generic[~ContextId, ~ReturnValue]): View Source

11class PickleFileCache(CacheBase):
12    """Cache the result as pickle files saved on the disk."""
13
14    def __init__(self, directory: os.PathLike = None, pickle_protocol: int = None) -> None:
15        """
16        Args:
17            directory: the directory where the files will be saved. The directory will be created if it does not exist.
18                       If None, use the global default `cache.filesystem.directory`
19            pickle_protocol: the protocol used when pickling files. If None, use the global default
20                             `cache.pickle_protocol`
21        """
22
23        self.__directory = pathlib.Path(directory if directory is not None else defaults["cache.filesystem.directory"])
24        self.__directory.mkdir(parents=True, exist_ok=True)
25
26        self.__pickle_protocol = pickle_protocol if pickle_protocol is not None else defaults["cache.pickle_protocol"]
27
28    def _get_file_path(self, context_id: str) -> pathlib.Path:
29        """
30        Args:
31            context_id: the file name without the file extension
32
33        Returns:
34            The full file path used in this cache that corresponds to the context id
35        """
36
37        filename = f"{context_id}.pickle"
38        return self.__directory.joinpath(filename)
39
40    def save(self, context_id: str, result: ReturnValue) -> None:
41        """
42        Save the result with the given context id.
43
44        Args:
45            context_id: identifier of the function call context, must be a valid file name without the file extension
46            result: return value of the function call
47        """
48
49        with open(self._get_file_path(context_id), mode="wb") as file:
50            pickle.dump(result, file, protocol=self.__pickle_protocol)
51
52    def retrieve(self, context_id: str) -> ReturnValue:
53        """
54        Retrieve the function return value with the given context id.
55        If there is no cached results for the context_id, throws a checkpointing.exceptions.CheckpointNotExist
56
57        Args:
58            context_id: identifier of the function call context, must be a valid file name without the file extension
59
60        Returns:
61            The return value of the function that corresponds to this context id
62        """
63
64        path = self._get_file_path(context_id)
65        if not path.exists():
66            raise CheckpointNotExist
67
68        with open(path, mode="rb") as file:
69            return pickle.load(file, protocol=self.__pickle_protocol)

Cache the result as pickle files saved on the disk.

PickleFileCache(directory: os.PathLike = None, pickle_protocol: int = None) View Source

14    def __init__(self, directory: os.PathLike = None, pickle_protocol: int = None) -> None:
15        """
16        Args:
17            directory: the directory where the files will be saved. The directory will be created if it does not exist.
18                       If None, use the global default `cache.filesystem.directory`
19            pickle_protocol: the protocol used when pickling files. If None, use the global default
20                             `cache.pickle_protocol`
21        """
22
23        self.__directory = pathlib.Path(directory if directory is not None else defaults["cache.filesystem.directory"])
24        self.__directory.mkdir(parents=True, exist_ok=True)
25
26        self.__pickle_protocol = pickle_protocol if pickle_protocol is not None else defaults["cache.pickle_protocol"]

Args

directory: the directory where the files will be saved. The directory will be created if it does not exist. If None, use the global default cache.filesystem.directory
pickle_protocol: the protocol used when pickling files. If None, use the global default cache.pickle_protocol

def save(self, context_id: str, result: ~ReturnValue) -> None: View Source

40    def save(self, context_id: str, result: ReturnValue) -> None:
41        """
42        Save the result with the given context id.
43
44        Args:
45            context_id: identifier of the function call context, must be a valid file name without the file extension
46            result: return value of the function call
47        """
48
49        with open(self._get_file_path(context_id), mode="wb") as file:
50            pickle.dump(result, file, protocol=self.__pickle_protocol)

Save the result with the given context id.

Args

context_id: identifier of the function call context, must be a valid file name without the file extension
result: return value of the function call

def retrieve(self, context_id: str) -> ~ReturnValue: View Source

52    def retrieve(self, context_id: str) -> ReturnValue:
53        """
54        Retrieve the function return value with the given context id.
55        If there is no cached results for the context_id, throws a checkpointing.exceptions.CheckpointNotExist
56
57        Args:
58            context_id: identifier of the function call context, must be a valid file name without the file extension
59
60        Returns:
61            The return value of the function that corresponds to this context id
62        """
63
64        path = self._get_file_path(context_id)
65        if not path.exists():
66            raise CheckpointNotExist
67
68        with open(path, mode="rb") as file:
69            return pickle.load(file, protocol=self.__pickle_protocol)

Retrieve the function return value with the given context id. If there is no cached results for the context_id, throws a checkpointing.exceptions.CheckpointNotExist

Args

context_id: identifier of the function call context, must be a valid file name without the file extension

Returns

The return value of the function that corresponds to this context id

Inherited Members

checkpointing.cache.base.CacheBase: synchronize_with