task-checkpoint 0.1.5

A small Python library for resumable task pools in long-running scripts.

Task Pool

Task Pool is a small Python library for long-running scripts that need to process many small jobs without losing progress.

Turn a fragile for loop into a resumable task pool.

When a run stops halfway, Task Pool remembers what is done, what is running, and what still needs work. You can restart the script and continue from the last known point.

Use it for crawling, experiments, data analysis, batch API calls, file processing, and other work where unfinished items should be easy to resume.

It feels like a tiny job queue, but stays inside your Python script. For local use, no database server, message queue, or service setup is needed.

Quickstart

Install it with:

pip install task-checkpoint

Turn a regular loop:

for url in urls:
    fetch_url(url)

into a resumable loop:

from task_pool import checkpoint


for url in checkpoint(urls, "urls"):
    fetch_url(url)

Progress is stored in .task_pool/urls.json. Completed items are skipped the next time the script runs. If processing raises an exception, the current item is rolled back to not_start.

Use a stable, unique name for each task set. You can also choose the state path directly:

for url in checkpoint(
    urls,
    state_path="state/urls.checkpoint.json",
):
    fetch_url(url)

The explicit class API is equivalent:

from task_pool import TaskPool


pool = TaskPool.from_iterable(
    urls,
    state_path="state/urls.checkpoint.json",
)

Structured payloads work too:

jobs = [
    {"input_path": "data/a.json", "method": "baseline"},
    {"input_path": "data/b.json", "method": "baseline"},
]

for job in checkpoint(jobs, "analysis-jobs"):
    run_analysis(job["input_path"], job["method"])

The iterable is fully consumed when the pool is created, so it must be finite and its payloads must be JSON-serializable. It is treated as the current task snapshot: existing items keep their progress, new items start as not_start, missing items are removed, and duplicate payloads become one task.

TaskPool API

Use TaskPool when your tasks already live in a file or store, or when you want to append jobs from Python.

Put one URL per line:

https://example.com/a
https://example.com/b
https://example.com/c

Then process the file as a task pool:

from task_pool import TaskPool


pool = TaskPool("urls.txt")

def fetch_one(url):
    fetch_url(url)


pool.for_each(fetch_one)

For very small scripts, a lambda also works:

pool.for_each(lambda url: fetch_url(url))

You can also use a regular loop:

for url in pool:
    fetch_url(url)

If you need task metadata, use iter_with_metadata():

for task in pool.iter_with_metadata():
    result = fetch_url(task.payload)
    save_result(task.key, result)

Each completed task is committed. If an exception is raised, the current task is rolled back to not_start and the exception is raised again.

You can print a small progress summary:

print(pool.stats())
# {"total": 10, "not_start": 3, "pending": 0, "committed": 7}

For structured jobs, use a JSON task pool:

pool = TaskPool("tasks.json")

pool.append({"input_path": "data/a.json", "method": "baseline"})
pool.append({"input_path": "data/b.json", "method": "baseline"})

Manual Control

Use lease() when you want one task at a time and need explicit control.

pool = TaskPool("tasks.json")

with pool.lease() as task:
    if task is None:
        return

    do_work(task.payload)

If there is no task to process, lease() returns None.

Leaving the with block normally commits the task. If an exception is raised, the task is rolled back automatically.

Store Selection

You can choose a store by file suffix:

TaskPool("tasks.json")      # JSONFileStore
TaskPool("tasks.sqlite")    # SQLiteStore
TaskPool("tasks.sqlite3")   # SQLiteStore
TaskPool("tasks.db")        # SQLiteStore
TaskPool("rows.csv")        # CSVRowStore with sidecar state
TaskPool("rows.tsv")        # CSVRowStore with sidecar state
TaskPool("urls.txt")        # TextLineStore with sidecar state

You can also pass a store explicitly:

from task_pool import JSONFileStore, SQLiteStore, TaskPool


pool = TaskPool(JSONFileStore("tasks.json"))
pool = TaskPool(SQLiteStore("tasks.db"))

Source Files

Task Pool can also read tasks from source files.

Text files

pool = TaskPool("urls.txt")

Example:

https://example.com/a
https://example.com/b
https://example.com/c

Each non-empty line becomes a string payload:

"https://example.com/a"

CSV and TSV files

pool = TaskPool("rows.csv")
pool = TaskPool("rows.tsv")

By default, CSV and TSV files are read with a header row:

id,name
job-1,First
job-2,Second

The first row becomes:

{"id": "job-1", "name": "First"}

For files without headers:

pool = TaskPool.csv("rows.csv", has_header=False)
pool = TaskPool.tsv("rows.tsv", has_header=False)

Rows are named col1, col2, and so on:

{"col1": "job-1", "col2": "First"}

Task keys are derived from canonical payload content, not row position. Duplicate payloads become one task, so inserting or reordering rows does not reset their progress.

Source files are not modified. A sidecar JSON file stores only progress state, for example:

rows.csv.task_pool.json
urls.txt.task_pool.json

When a source-backed pool opens, it synchronizes the sidecar with the current source file. New payloads start as not_start, existing payloads keep their status, and payloads no longer in the source are removed.

Do not change a source file while workers are running.

Task Status

Each task has one of these statuses:

• not_start
• pending
• committed

The common flow is:

not_start -> pending -> committed
                  |
                  -> not_start

pending means a worker has leased the task and is currently processing it.

Recovering Interrupted Tasks

Exceptions and KeyboardInterrupt roll back the current task automatically.

A power outage or forced process termination may stop Python before rollback runs. The current task will then remain in pending.

When no previous workers are still running, reset all pending tasks before starting again:

pool.reset_pending()

If some workers may still be running, reset only tasks that have remained pending longer than a chosen period:

pool.reset_pending_older_than(age_seconds=3600)

Choose a period longer than the expected processing time of one task.

Notes

• The default store is JSONFileStore("task_pool.json").
• Payloads are stored as JSON.
• Task Pool provides at-least-once processing. If work finishes but the process stops before commit, the task may run again. Writes outside Task Pool should be idempotent.
• JSON and source-backed stores use file locks and atomic writes.
• SQLite is better for larger pools or heavier concurrent use.
• Source-backed stores keep the source file unchanged and write progress to a sidecar JSON file.

Development

Run the test suite with:

python -m unittest