Checkpointer

A lightweight library for application checkpointing in Python. Checkpoints allow an application to skip expensive or redundant computation by remembering that a step has already completed.

Installation

Install directly from GitHub:

pip install git+https://github.com/milesbarr/checkpointer@main

Or clone the repository and install locally:

git clone https://github.com/milesbarr/checkpointer.git
cd checkpointer
pip install .

Usage

Store checkpoints in a file with shelve:

importshelvefromcheckpointerimportcheckpointwithshelve.open("checkpoints") ascheckpoints:
withcheckpoint(checkpoints, "my-task", version=1) ascp:
# Skip work if this checkpoint already exists.cp.skip_if_exists()
# Expensive work goes here...
...
# The checkpoint will be recorded automatically when the block exits.

Store checkpoints in a SQLite database:

fromcheckpointerimportcheckpointimportsqlite3con=sqlite3.connect(":memory:")
withcheckpoint(con, "my-task", version=1) ascp:
# Skip work if this checkpoint already exists.cp.skip_if_exists()
# Expensive work goes here.
...
# The checkpoint will be recorded automatically when the block exits.

Example

Process files in a directory only if they have changed since the last checkpoint:

fromcheckpointerimportcheckpointfrompathlibimportPathimportshelvedirectory=Path("data")
withshelve.open("checkpoints") ascheckpoints:
forfileindirectory.glob("*.txt"):
withcheckpoint(checkpoints, file, file.stat().st_mtime) ascp:
cp.skip_if_exists()
process_file(file)

License

This project is licensed under the MIT License.