pykitchen 0.1.1

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. Runtime dependency: pydantic.

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

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
Task	Base task model — subclass it and implement process()
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 task.process() for each task
BaseQueue / BaseStorage	ABCs for custom backends (SQS, S3, Azure, …)

Quickstart

Define your task by subclassing Task and implementing process():

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

class PrintTask(Task):
    message: str

    def process(self) -> list:
        print(f"[{self.task_id}] {self.message}")
        return []  # return follow-up Task instances to enqueue them

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

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

q.enqueue(PrintTask(message="hello"))

run_worker(worker_cfg, q, stop)

process() must raise on failure (the worker handles retry/dead-letter) and be idempotent (delivery is at-least-once). Return a list of Task instances from process() to enqueue follow-up work.

Manual inbox

Drop a JSON file into data/manual_inbox/ — it's picked up automatically on the next worker loop iteration. The file must include a __type__ field matching your Task subclass name:

echo '{"__type__": "PrintTask", "message": "hi"}' > 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.

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 base 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 (35 tests)
pyproject.toml                  Build metadata (hatchling, UV-compatible)

Running tests

uv run pytest