# checkpointer · [](https://github.com/Reddan/checkpointer/blob/master/LICENSE) [](https://pypi.org/project/checkpointer/) [](https://pypi.org/project/checkpointer/)
`checkpointer` is a Python library offering a decorator-based API for memoizing (caching) function results with code-aware cache invalidation. It works with sync and async functions, supports multiple storage backends, and invalidates caches automatically when your code or dependencies change - helping you maintain correctness, speed up execution, and smooth out your workflows by skipping redundant, costly operations.
## 📦 Installation
```bash
pip install checkpointer
```
## 🚀 Quick Start
Apply the `@checkpoint` decorator to any function:
```python
from checkpointer import checkpoint
@checkpoint
def expensive_function(x: int) -> int:
print("Computing...")
return x ** 2
result = expensive_function(4) # Computes and stores the result
result = expensive_function(4) # Loads from the cache
```
## 🧠 How It Works
When you decorate a function with `@checkpoint` and call it, `checkpointer` computes a unique identifier that represents that specific call. This identifier is based on:
* The function's source code and all its user-defined dependencies,
* Global variables used by the function (if capturing is enabled or explicitly annotated),
* The actual arguments passed to the function.
`checkpointer` then looks up this identifier in its cache. If a valid cached result exists, it returns that immediately. Otherwise, it runs the original function, stores the result, and returns it.
`checkpointer` is designed to be flexible through features like:
* **Support for decorated methods**, correctly caching results bound to instances.
* **Support for decorated async functions**, compatible with any async runtime.
* **Robust hashing**, covering complex Python objects and large **NumPy**/**PyTorch** arrays via its internal `ObjectHash`.
* **Targeted hashing**, allowing you to optimize how arguments and captured variables are hashed.
* **Multi-layered caching**, letting you stack decorators for layered caching strategies without losing cache consistency.
### 🚨 What Causes Cache Invalidation?
To ensure cache correctness, `checkpointer` tracks two types of hashes:
#### 1. Function Identity Hash (Computed Once Per Function)
This hash represents the decorated function itself and is computed once (usually on first invocation). It covers:
* **Function Code and Signature:**\
The actual logic and parameters of the function are hashed - but *not* parameter type annotations or formatting details like whitespace, newlines, comments, or trailing commas, which do **not** trigger invalidation.
* **Dependencies:**\
All user-defined functions and methods that the decorated function calls or relies on, including indirect dependencies, are included recursively. Dependencies are identified by:
* Inspecting the function's global scope for referenced functions and objects.
* Inferring from the function's argument type annotations.
* Analyzing object constructions and method calls to identify classes and methods used.
* **Exclusions:**\
Changes elsewhere in the module unrelated to the function or its dependencies do **not** cause invalidation.
#### 2. Call Hash (Computed on Every Function Call)
Every function call produces a call hash, combining:
* **Passed Arguments:**\
Includes positional and keyword arguments, combined with default values. Changing defaults alone doesn't necessarily trigger invalidation unless it affects actual call values.
* **Captured Global Variables:**\
When `capture=True` or explicit capture annotations are used, `checkpointer` includes referenced global variables in the call hash. Variables annotated with `CaptureMe` are hashed on every call, causing immediate cache invalidation if they change. Variables annotated with `CaptureMeOnce` are hashed only once per Python session, improving performance by avoiding repeated hashing.
* **Custom Argument Hashing:**\
Using `HashBy` annotations, arguments or captured variables can be transformed before hashing (e.g., sorting lists to ignore order), allowing more precise or efficient call hashes.
## 💡 Usage
Once a function is decorated with `@checkpoint`, you can interact with its caching behavior using the following methods:
* **`expensive_function(...)`**:\
Call the function normally. This will compute and cache the result or load it from cache.
* **`expensive_function.rerun(...)`**:\
Force the original function to execute and overwrite any existing cached result.
* **`expensive_function.fn(...)`**:\
Call the undecorated function directly, bypassing the cache (useful in recursion to prevent caching intermediate steps).
* **`expensive_function.get(...)`**:\
Retrieve the cached result without executing the function. Raises `CheckpointError` if no valid cache exists.
* **`expensive_function.exists(...)`**:\
Check if a cached result exists without computing or loading it.
* **`expensive_function.delete(...)`**:\
Remove the cached entry for given arguments.
* **`expensive_function.reinit(recursive: bool = True)`**:\
Recalculate the function identity hash and recapture `CaptureMeOnce` variables, updating the cached function state within the same Python session.
## ⚙️ Configuration & Customization
The `@checkpoint` decorator accepts the following parameters:
* **`storage`** (Type: `str` or `checkpointer.Storage`, Default: `"pickle"`)\
Storage backend to use: `"pickle"` (disk-based, persistent), `"memory"` (in-memory, non-persistent), or a custom `Storage` class.
* **`directory`** (Type: `str` or `pathlib.Path` or `None`, Default: `~/.cache/checkpoints`)\
Base directory for disk-based checkpoints (only for `"pickle"` storage).
* **`capture`** (Type: `bool`, Default: `False`)\
If `True`, includes global variables referenced by the function in call hashes (except those excluded via `NoHash`).
* **`expiry`** (Type: `Callable[[datetime.datetime], bool]` or `datetime.timedelta`, Default: `None`)\
A custom callable that receives the `datetime` timestamp of a cached result. It should return `True` if the cached result is considered expired and needs recomputation, or `False` otherwise.
* **`fn_hash_from`** (Type: `Any`, Default: `None`)\
Override the computed function identity hash with any hashable object you provide (e.g., version strings, config IDs). This gives you explicit control over the function's version and when its cache should be invalidated.
* **`when`** (Type: `bool`, Default: `True`)\
Enable or disable checkpointing dynamically, useful for environment-based toggling.
* **`verbosity`** (Type: `int` (`0`, `1`, or `2`), Default: `1`)\
Controls the level of logging output from `checkpointer`.
* `0`: No output.
* `1`: Shows when functions are computed and cached.
* `2`: Also shows when cached results are remembered (loaded from cache).
## 🔬 Customize Argument Hashing
You can customize how arguments are hashed without modifying the actual argument values to improve cache hit rates or speed up hashing.
* **`Annotated[T, HashBy[fn]]`**:\
Transform the argument via `fn(argument)` before hashing. Useful for normalization (e.g., sorting lists) or optimized hashing for complex inputs.
* **`NoHash[T]`**:\
Exclude the argument from hashing completely, so changes to it won't trigger cache invalidation.
**Example:**
```python
from typing import Annotated
from checkpointer import checkpoint, HashBy, NoHash
from pathlib import Path
import logging
def file_bytes(path: Path) -> bytes:
return path.read_bytes()
@checkpoint
def process(
numbers: Annotated[list[int], HashBy[sorted]], # Hash by sorted list
data_file: Annotated[Path, HashBy[file_bytes]], # Hash by file content
log: NoHash[logging.Logger], # Exclude logger from hashing
):
...
```
In this example, the hash for `numbers` ignores order, `data_file` is hashed based on its contents rather than path, and changes to `log` don't affect caching.
## 🎯 Capturing Global Variables
`checkpointer` can include **captured global variables** in call hashes - these are globals your function reads during execution that may affect results.
Use `capture=True` on `@checkpoint` to capture **all** referenced globals (except those explicitly excluded with `NoHash`).
Alternatively, you can **opt-in selectively** by annotating globals with:
* **`CaptureMe[T]`**:\
Capture the variable on every call (triggers invalidation on changes).
* **`CaptureMeOnce[T]`**:\
Capture once per Python session (for expensive, immutable globals).
You can also combine these with `HashBy` to customize how captured variables are hashed (e.g., hash by subset of attributes).
**Example:**
```python
from typing import Annotated
from checkpointer import checkpoint, CaptureMe, CaptureMeOnce, HashBy
from pathlib import Path
def file_bytes(path: Path) -> bytes:
return path.read_bytes()
captured_data: CaptureMe[Annotated[Path, HashBy[file_bytes]]] = Path("data.txt")
session_config: CaptureMeOnce[dict] = {"mode": "prod"}
@checkpoint
def process():
# `captured_data` is included in the call hash on every call, hashed by file content
# `session_config` is hashed once per session
...
```
## 🗄️ Custom Storage Backends
Implement your own storage backend by subclassing `checkpointer.Storage` and overriding required methods.
Within storage methods, `call_hash` identifies calls by arguments. Use `self.fn_id()` to get function identity (name + hash/version), important for organizing checkpoints.
**Example:**
```python
from checkpointer import checkpoint, Storage
from datetime import datetime
class MyCustomStorage(Storage):
def exists(self, call_hash):
fn_dir = self.checkpointer.directory / self.fn_id()
return (fn_dir / call_hash).exists()
def store(self, call_hash, data):
... # Store serialized data
return data # Must return data to checkpointer
def checkpoint_date(self, call_hash): ...
def load(self, call_hash): ...
def delete(self, call_hash): ...
@checkpoint(storage=MyCustomStorage)
def custom_cached_function(x: int):
return x ** 2
```
Raw data
{
"_id": null,
"home_page": null,
"name": "checkpointer",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "async, cache, caching, data analysis, data processing, fast, hashing, invalidation, memoization, optimization, performance, workflow",
"author": "Hampus Hallman",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/fc/62/aaa5722f05bd690e24c9e6db1a5ba8b09c0585d0c4e16ac439f8f88cba9e/checkpointer-2.14.2.tar.gz",
"platform": null,
"description": "# checkpointer \u00b7 [](https://github.com/Reddan/checkpointer/blob/master/LICENSE) [](https://pypi.org/project/checkpointer/) [](https://pypi.org/project/checkpointer/)\n\n`checkpointer` is a Python library offering a decorator-based API for memoizing (caching) function results with code-aware cache invalidation. It works with sync and async functions, supports multiple storage backends, and invalidates caches automatically when your code or dependencies change - helping you maintain correctness, speed up execution, and smooth out your workflows by skipping redundant, costly operations.\n\n## \ud83d\udce6 Installation\n\n```bash\npip install checkpointer\n```\n\n## \ud83d\ude80 Quick Start\n\nApply the `@checkpoint` decorator to any function:\n\n```python\nfrom checkpointer import checkpoint\n\n@checkpoint\ndef expensive_function(x: int) -> int:\n print(\"Computing...\")\n return x ** 2\n\nresult = expensive_function(4) # Computes and stores the result\nresult = expensive_function(4) # Loads from the cache\n```\n\n## \ud83e\udde0 How It Works\n\nWhen you decorate a function with `@checkpoint` and call it, `checkpointer` computes a unique identifier that represents that specific call. This identifier is based on:\n\n* The function's source code and all its user-defined dependencies,\n* Global variables used by the function (if capturing is enabled or explicitly annotated),\n* The actual arguments passed to the function.\n\n`checkpointer` then looks up this identifier in its cache. If a valid cached result exists, it returns that immediately. Otherwise, it runs the original function, stores the result, and returns it.\n\n`checkpointer` is designed to be flexible through features like:\n\n* **Support for decorated methods**, correctly caching results bound to instances.\n* **Support for decorated async functions**, compatible with any async runtime.\n* **Robust hashing**, covering complex Python objects and large **NumPy**/**PyTorch** arrays via its internal `ObjectHash`.\n* **Targeted hashing**, allowing you to optimize how arguments and captured variables are hashed.\n* **Multi-layered caching**, letting you stack decorators for layered caching strategies without losing cache consistency.\n\n### \ud83d\udea8 What Causes Cache Invalidation?\n\nTo ensure cache correctness, `checkpointer` tracks two types of hashes:\n\n#### 1. Function Identity Hash (Computed Once Per Function)\n\nThis hash represents the decorated function itself and is computed once (usually on first invocation). It covers:\n\n* **Function Code and Signature:**\\\n The actual logic and parameters of the function are hashed - but *not* parameter type annotations or formatting details like whitespace, newlines, comments, or trailing commas, which do **not** trigger invalidation.\n\n* **Dependencies:**\\\n All user-defined functions and methods that the decorated function calls or relies on, including indirect dependencies, are included recursively. Dependencies are identified by:\n * Inspecting the function's global scope for referenced functions and objects.\n * Inferring from the function's argument type annotations.\n * Analyzing object constructions and method calls to identify classes and methods used.\n\n* **Exclusions:**\\\n Changes elsewhere in the module unrelated to the function or its dependencies do **not** cause invalidation.\n\n#### 2. Call Hash (Computed on Every Function Call)\n\nEvery function call produces a call hash, combining:\n\n* **Passed Arguments:**\\\n Includes positional and keyword arguments, combined with default values. Changing defaults alone doesn't necessarily trigger invalidation unless it affects actual call values.\n\n* **Captured Global Variables:**\\\n When `capture=True` or explicit capture annotations are used, `checkpointer` includes referenced global variables in the call hash. Variables annotated with `CaptureMe` are hashed on every call, causing immediate cache invalidation if they change. Variables annotated with `CaptureMeOnce` are hashed only once per Python session, improving performance by avoiding repeated hashing.\n\n* **Custom Argument Hashing:**\\\n Using `HashBy` annotations, arguments or captured variables can be transformed before hashing (e.g., sorting lists to ignore order), allowing more precise or efficient call hashes.\n\n## \ud83d\udca1 Usage\n\nOnce a function is decorated with `@checkpoint`, you can interact with its caching behavior using the following methods:\n\n* **`expensive_function(...)`**:\\\n Call the function normally. This will compute and cache the result or load it from cache.\n\n* **`expensive_function.rerun(...)`**:\\\n Force the original function to execute and overwrite any existing cached result.\n\n* **`expensive_function.fn(...)`**:\\\n Call the undecorated function directly, bypassing the cache (useful in recursion to prevent caching intermediate steps).\n\n* **`expensive_function.get(...)`**:\\\n Retrieve the cached result without executing the function. Raises `CheckpointError` if no valid cache exists.\n\n* **`expensive_function.exists(...)`**:\\\n Check if a cached result exists without computing or loading it.\n\n* **`expensive_function.delete(...)`**:\\\n Remove the cached entry for given arguments.\n\n* **`expensive_function.reinit(recursive: bool = True)`**:\\\n Recalculate the function identity hash and recapture `CaptureMeOnce` variables, updating the cached function state within the same Python session.\n\n## \u2699\ufe0f Configuration & Customization\n\nThe `@checkpoint` decorator accepts the following parameters:\n\n* **`storage`** (Type: `str` or `checkpointer.Storage`, Default: `\"pickle\"`)\\\n Storage backend to use: `\"pickle\"` (disk-based, persistent), `\"memory\"` (in-memory, non-persistent), or a custom `Storage` class.\n\n* **`directory`** (Type: `str` or `pathlib.Path` or `None`, Default: `~/.cache/checkpoints`)\\\n Base directory for disk-based checkpoints (only for `\"pickle\"` storage).\n\n* **`capture`** (Type: `bool`, Default: `False`)\\\n If `True`, includes global variables referenced by the function in call hashes (except those excluded via `NoHash`).\n\n* **`expiry`** (Type: `Callable[[datetime.datetime], bool]` or `datetime.timedelta`, Default: `None`)\\\n A custom callable that receives the `datetime` timestamp of a cached result. It should return `True` if the cached result is considered expired and needs recomputation, or `False` otherwise.\n\n* **`fn_hash_from`** (Type: `Any`, Default: `None`)\\\n Override the computed function identity hash with any hashable object you provide (e.g., version strings, config IDs). This gives you explicit control over the function's version and when its cache should be invalidated.\n\n* **`when`** (Type: `bool`, Default: `True`)\\\n Enable or disable checkpointing dynamically, useful for environment-based toggling.\n\n* **`verbosity`** (Type: `int` (`0`, `1`, or `2`), Default: `1`)\\\n Controls the level of logging output from `checkpointer`.\n * `0`: No output.\n * `1`: Shows when functions are computed and cached.\n * `2`: Also shows when cached results are remembered (loaded from cache).\n\n## \ud83d\udd2c Customize Argument Hashing\n\nYou can customize how arguments are hashed without modifying the actual argument values to improve cache hit rates or speed up hashing.\n\n* **`Annotated[T, HashBy[fn]]`**:\\\n Transform the argument via `fn(argument)` before hashing. Useful for normalization (e.g., sorting lists) or optimized hashing for complex inputs.\n\n* **`NoHash[T]`**:\\\n Exclude the argument from hashing completely, so changes to it won't trigger cache invalidation.\n\n**Example:**\n\n```python\nfrom typing import Annotated\nfrom checkpointer import checkpoint, HashBy, NoHash\nfrom pathlib import Path\nimport logging\n\ndef file_bytes(path: Path) -> bytes:\n return path.read_bytes()\n\n@checkpoint\ndef process(\n numbers: Annotated[list[int], HashBy[sorted]], # Hash by sorted list\n data_file: Annotated[Path, HashBy[file_bytes]], # Hash by file content\n log: NoHash[logging.Logger], # Exclude logger from hashing\n):\n ...\n```\n\nIn this example, the hash for `numbers` ignores order, `data_file` is hashed based on its contents rather than path, and changes to `log` don't affect caching.\n\n## \ud83c\udfaf Capturing Global Variables\n\n`checkpointer` can include **captured global variables** in call hashes - these are globals your function reads during execution that may affect results.\n\nUse `capture=True` on `@checkpoint` to capture **all** referenced globals (except those explicitly excluded with `NoHash`).\n\nAlternatively, you can **opt-in selectively** by annotating globals with:\n\n* **`CaptureMe[T]`**:\\\n Capture the variable on every call (triggers invalidation on changes).\n\n* **`CaptureMeOnce[T]`**:\\\n Capture once per Python session (for expensive, immutable globals).\n\nYou can also combine these with `HashBy` to customize how captured variables are hashed (e.g., hash by subset of attributes).\n\n**Example:**\n\n```python\nfrom typing import Annotated\nfrom checkpointer import checkpoint, CaptureMe, CaptureMeOnce, HashBy\nfrom pathlib import Path\n\ndef file_bytes(path: Path) -> bytes:\n return path.read_bytes()\n\ncaptured_data: CaptureMe[Annotated[Path, HashBy[file_bytes]]] = Path(\"data.txt\")\nsession_config: CaptureMeOnce[dict] = {\"mode\": \"prod\"}\n\n@checkpoint\ndef process():\n # `captured_data` is included in the call hash on every call, hashed by file content\n # `session_config` is hashed once per session\n ...\n```\n\n## \ud83d\uddc4\ufe0f Custom Storage Backends\n\nImplement your own storage backend by subclassing `checkpointer.Storage` and overriding required methods.\n\nWithin storage methods, `call_hash` identifies calls by arguments. Use `self.fn_id()` to get function identity (name + hash/version), important for organizing checkpoints.\n\n**Example:**\n\n```python\nfrom checkpointer import checkpoint, Storage\nfrom datetime import datetime\n\nclass MyCustomStorage(Storage):\n def exists(self, call_hash):\n fn_dir = self.checkpointer.directory / self.fn_id()\n return (fn_dir / call_hash).exists()\n\n def store(self, call_hash, data):\n ... # Store serialized data\n return data # Must return data to checkpointer\n\n def checkpoint_date(self, call_hash): ...\n def load(self, call_hash): ...\n def delete(self, call_hash): ...\n\n@checkpoint(storage=MyCustomStorage)\ndef custom_cached_function(x: int):\n return x ** 2\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "checkpointer adds code-aware caching to Python functions, maintaining correctness and speeding up execution as your code changes.",
"version": "2.14.2",
"project_urls": {
"Repository": "https://github.com/Reddan/checkpointer.git"
},
"split_keywords": [
"async",
" cache",
" caching",
" data analysis",
" data processing",
" fast",
" hashing",
" invalidation",
" memoization",
" optimization",
" performance",
" workflow"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "f73b93b1774a4af077616f082badd98bce9c3ba7a001b0ae93df8eaeb2775a84",
"md5": "54eac25d55a3e1a0f09857c39d876124",
"sha256": "9078d7e59ee9bedab477434a742f285a15340ef088d535b755cec653713c4191"
},
"downloads": -1,
"filename": "checkpointer-2.14.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "54eac25d55a3e1a0f09857c39d876124",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 22638,
"upload_time": "2025-07-09T11:26:03",
"upload_time_iso_8601": "2025-07-09T11:26:03.352199Z",
"url": "https://files.pythonhosted.org/packages/f7/3b/93b1774a4af077616f082badd98bce9c3ba7a001b0ae93df8eaeb2775a84/checkpointer-2.14.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "fc62aaa5722f05bd690e24c9e6db1a5ba8b09c0585d0c4e16ac439f8f88cba9e",
"md5": "532c8308a6b295ea9011d11ee397b96d",
"sha256": "18d249b7925221f52b5de70207866f3e2289d3e1efe9c92911093bab0779a5af"
},
"downloads": -1,
"filename": "checkpointer-2.14.2.tar.gz",
"has_sig": false,
"md5_digest": "532c8308a6b295ea9011d11ee397b96d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 38194,
"upload_time": "2025-07-09T11:26:05",
"upload_time_iso_8601": "2025-07-09T11:26:05.418141Z",
"url": "https://files.pythonhosted.org/packages/fc/62/aaa5722f05bd690e24c9e6db1a5ba8b09c0585d0c4e16ac439f8f88cba9e/checkpointer-2.14.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-09 11:26:05",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Reddan",
"github_project": "checkpointer",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "checkpointer"
}
JSON
