Python

Python SDK

The Python SDK provides a clean interface for writing TaskDaemon handlers.

Installation

pip install taskdaemon

Basic Usage

from taskdaemon import run, Task, Success, Error

def handler(task: Task) -> Success | Error:
    name = task.task_data.get("name", "World")
    return Success({"message": f"Hello, {name}!"})

run(handler)

Task Object

The Task dataclass contains:

@dataclass
class Task:
    task_id: str      # Unique task identifier
    task_type: str    # Handler type (e.g., "resize")
    task_data: dict   # Input data from client
    attempt: int      # Current attempt number (starts at 0)

Return Types

Success

Return Success with a result (any JSON-serializable value):

return Success({"processed": True, "count": 42})

Error

Return Error with an error message and optional retryable flag:

# Non-retryable error (default)
return Error("Invalid input: missing required field")

# Retryable error (will be retried)
return Error("Service temporarily unavailable", retryable=True)

Complete Example

Here’s the actual image handler from the sample project:

import base64
from io import BytesIO
import requests
from PIL import Image
from taskdaemon import run, Task, Success, Error

def handler(task: Task) -> Success | Error:
    data = task.task_data
    
    # Download image
    resp = requests.get(data["image_url"], timeout=30)
    img = Image.open(BytesIO(resp.content))
    
    if task.task_type == "resize":
        img = img.resize((data["width"], data["height"]), Image.LANCZOS)
    elif task.task_type == "thumbnail":
        size = data["size"]
        img.thumbnail((size, size), Image.LANCZOS)
    
    # Encode as base64
    buffer = BytesIO()
    img.save(buffer, format="PNG")
    b64 = base64.b64encode(buffer.getvalue()).decode()
    
    return Success({
        "size": [img.width, img.height],
        "format": "png",
        "data": b64
    })

if __name__ == "__main__":
    run(handler)

Error Handling

Exceptions are automatically caught and converted to non-retryable errors:

def handler(task: Task) -> Success | Error:
    # If this raises, it becomes Error(str(exception), retryable=False)
    value = task.task_data["required_field"]
    return Success({"value": value})

For explicit error handling:

def handler(task: Task) -> Success | Error:
    if "required_field" not in task.task_data:
        return Error("Missing required_field")
    
    try:
        result = external_api_call()
        return Success(result)
    except ConnectionError:
        return Error("API unavailable", retryable=True)
    except ValueError as e:
        return Error(f"Invalid data: {e}")

Dockerfile

FROM python:3.11-slim

RUN pip install taskdaemon pillow requests

COPY handler.py /app/
WORKDIR /app

CMD ["python", "-u", "handler.py"]

:::tip Use python -u for unbuffered output to ensure responses are sent immediately. :::

Handler Configuration

[handlers.resize]
image = "image-handler:latest"
instances = 4
timeout = 30

[handlers.thumbnail]
image = "image-handler:latest"
instances = 2
timeout = 15

Multiple task types can use the same image - the handler receives task.task_type to differentiate.