pykitchen 0.1.0

Building blocks for single-consumer async task queue processing

pykitchen

SDK building blocks for single-consumer async task queue processing in Python.

Provides a local-filesystem queue with deferred retries, a manual priority lane, and pluggable storage/queue backend interfaces — with zero runtime dependencies.

examples/kafka_spark_service/ shows how to wire the SDK into a service that ingests from Kafka via Spark Structured Streaming, with one-at-a-time sequential processing.

How it works

The worker drains a queue one task at a time. On failure a task is deferred to data/retry/ with its due time encoded in the filename — so the retry schedule survives restarts without any in-memory timer. The manual inbox and a priority lane let you inject or fast-track tasks without touching Kafka.

Kafka --Spark--> enqueue ----+
manual drop  --> enqueue ----+--> data/pending/ --> in-memory queue --> worker --> data/processed/
                              |                                            |
data/priority/<id> -----------> served ahead of queue                    +-------> data/retry/  (deferred)
                                                                         \-------> data/failed/ (dead letter)

Install

# Core SDK — no runtime dependencies
pip install pykitchen

# With Spark support (requires Java 8 or 11)
pip install "pykitchen[spark]"

Or with uv:

uv sync --group dev            # SDK + dev tools (pytest, ruff)
uv sync --extra spark --group dev  # also installs pyspark==3.1.3

SDK building blocks

Building block	What it gives you
LocalConfig	Filesystem path layout (pending/, retry/, failed/, …) + ensure_dirs()
WorkerConfig	Retry and timing settings (max_retries, retry_delay_seconds, …)
LocalStorage	Full task lifecycle on disk (persist, ack, nack, retry, dead-letter, priority)
LocalQueue	Thread-safe in-memory queue backed by LocalStorage
run_worker()	Single-consumer loop — calls your process(task) function
Task	Shared task model (payload, task_id, source, attempts, …)
BaseQueue / BaseStorage	ABCs for custom backends (SQS, S3, Azure, …)

Quickstart

import threading
from pykitchen import LocalConfig, LocalQueue, LocalStorage, WorkerConfig, run_worker
from pykitchen.task import Task

cfg = LocalConfig()          # uses ./data by default
cfg.ensure_dirs()

q = LocalQueue(cfg, LocalStorage(cfg))
worker_cfg = WorkerConfig(max_retries=3, retry_delay_seconds=1800)
stop = threading.Event()

def process(task: Task) -> None:
    print(f"Processing {task.task_id}: {task.payload}")
    # raise an exception to trigger retry / dead-letter

# Inject a message
q.enqueue({"id": 1, "msg": "hello"}, source="manual")

run_worker(worker_cfg, q, process, stop)

Manual inbox

Drop a JSON file into data/manual_inbox/ — it's picked up automatically:

echo '{"id": 1, "msg": "hello"}' > data/manual_inbox/a.json

Prioritise a queued task

Pending files are named <task_id>.json. To jump one to the front of the queue:

touch data/priority/<task_id>

Kafka + Spark example

examples/kafka_spark_service/ is a complete runnable service:

# No-Kafka mode — manual inbox only
ENABLE_STREAM=0 uv run python -m examples.kafka_spark_service.main

# With Kafka + Spark
ENABLE_STREAM=1 KAFKA_BOOTSTRAP=localhost:9092 KAFKA_TOPIC=input-topic \
  uv run python -m examples.kafka_spark_service.main

Put your business logic in examples/kafka_spark_service/processor.py. The process() function must raise on failure (the worker handles retry/dead-letter) and be idempotent (delivery is at-least-once).

Configuration (kafka_spark_service example)

Variable	Default	Meaning
ENABLE_STREAM	true	0 = manual-inbox-only, no Kafka/Spark
KAFKA_BOOTSTRAP	localhost:9092	Kafka brokers
KAFKA_TOPIC	input-topic	Source topic
DATA_DIR	./data	Root of all on-disk state
MAX_RETRIES	3	Max total attempts (initial + retries)
RETRY_DELAY_SECONDS	1800	Deferred-retry delay (seconds)
RETRY_BACKOFF_EXP	false	Exponential backoff on retries
QUEUE_GET_TIMEOUT	2.0	Idle poll cadence (seconds)

Project layout

src/pykitchen/                  SDK package
  task.py                       Task model
  worker.py                     WorkerConfig + run_worker()
  backends/
    local_config.py             LocalConfig (filesystem paths)
    local_queue.py              LocalQueue
    local_storage.py            LocalStorage
    base_queue.py / base_storage.py  ABCs for custom backends
    sqs_queue.py / s3_storage.py / azure_blob_storage.py  stubs

examples/kafka_spark_service/   Example service (Kafka + Spark)
  main.py                       Wiring: queue → stream → worker
  config.py                     KafkaSparkConfig (LocalConfig + WorkerConfig + Kafka/Spark)
  ingestion.py                  Spark Structured Streaming ingestion
  processor.py                  Business logic (replace with your own)

tests/                          pytest suite (31 tests)
pyproject.toml                  Build metadata (hatchling, UV-compatible)

Running tests

pytest