Skip to main content

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

Project description

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

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)

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():
    url = task.payload
    result = fetch_url(url)
    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.

For very small scripts, a lambda also works:

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

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

task_checkpoint-0.1.4.tar.gz (14.6 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

task_checkpoint-0.1.4-py3-none-any.whl (15.4 kB view details)

Uploaded Python 3

File details

Details for the file task_checkpoint-0.1.4.tar.gz.

File metadata

  • Download URL: task_checkpoint-0.1.4.tar.gz
  • Upload date:
  • Size: 14.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.14.0

File hashes

Hashes for task_checkpoint-0.1.4.tar.gz
Algorithm Hash digest
SHA256 b4a7c629e3f1f3104dd51bfecf9f02880962bf62772c91e13f77e99807e3266b
MD5 f79abf2fddc0b7d60bef5c1d9298f37a
BLAKE2b-256 8d9e6cdf1927a6fcb84a70ca8272492e25158929d7d055fd1c1442622a94fedb

See more details on using hashes here.

File details

Details for the file task_checkpoint-0.1.4-py3-none-any.whl.

File metadata

File hashes

Hashes for task_checkpoint-0.1.4-py3-none-any.whl
Algorithm Hash digest
SHA256 0f6a8f9a7d7604c905f7fd37dfd4b431bc529937fe775795d2cd7d3c4578c5fb
MD5 869d2e25a91932890555e123ed5ee938
BLAKE2b-256 32bfe911421d8ca52917a5bca48616563aefbbc1f4d3e1a68606bd12d23d9265

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page