Metadata-Version: 2.4
Name: pykitchen
Version: 0.1.1
Summary: Building blocks for single-consumer async task queue processing
Requires-Python: >=3.8
Requires-Dist: pydantic-settings>=2.0
Requires-Dist: pydantic>=2.0
Provides-Extra: spark
Requires-Dist: pyspark==3.1.3; extra == 'spark'
Description-Content-Type: text/markdown

# 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](https://docs.pydantic.dev/).

`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

```bash
pip install pykitchen

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

Or with [uv](https://docs.astral.sh/uv/):

```bash
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()`:

```python
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:

```bash
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:

```bash
touch data/priority/<task_id>
```

## Kafka + Spark example

`examples/kafka_spark_service/` is a complete runnable service:

```bash
# 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

```bash
uv run pytest
```
