Tasks¶

Tasks are Python functions decorated with @task that can be executed by FastWorker workers.

Defining Tasks¶

Use the @task decorator to register a function as a task:

from fastworker import task

@task
def add(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y

@task
def process_data(data: dict) -> dict:
    """Process some data."""
    return {"processed": True, "input": data}

@task
async def async_task(url: str) -> str:
    """Async task example."""
    # Async I/O operations
    return f"Fetched: {url}"

Task Requirements¶

Decorated with @task: Required for registration
Importable module: Tasks must be in a module that can be imported
Serializable arguments: Arguments must be JSON-serializable (or Pickle-serializable if using Pickle format)
Serializable return value: Return values must be serializable

Loading Tasks¶

Workers load tasks from modules specified with --task-modules:

# Load from single module
fastworker control-plane --task-modules mytasks

# Load from multiple modules (comma-separated)
fastworker control-plane --task-modules mytasks,other_tasks,more_tasks

Task Priority¶

Tasks can be submitted with different priority levels:

from fastworker.tasks.models import TaskPriority

# Using enum
task_id = await client.delay("my_task", priority=TaskPriority.CRITICAL)
task_id = await client.delay("my_task", priority=TaskPriority.HIGH)
task_id = await client.delay("my_task", priority=TaskPriority.NORMAL)
task_id = await client.delay("my_task", priority=TaskPriority.LOW)

# Using string
task_id = await client.delay("my_task", priority="high")

Priority Levels¶

Priority	Value	Description
`CRITICAL`	0	Highest priority, processed first
`HIGH`	1	High priority
`NORMAL`	2	Default priority
`LOW`	3	Lowest priority, processed last

Task State Machine¶

FastWorker uses a formal state machine to manage each task's lifecycle. States transition atomically under an asyncio.Lock, ensuring safe concurrent operations.

PENDING → QUEUED/SCHEDULED → ASSIGNED → RUNNING → SUCCESS | FAILURE → RETRYING
         ↓                                             ↓
      CANCELLED                                    CANCELLED

Task Status¶

Status	Description
`PENDING`	Task created, not yet queued
`SCHEDULED`	Task delayed with ETA, waiting for its execution time
`QUEUED`	Task in queue, waiting to be assigned
`ASSIGNED`	Task assigned to a worker
`RUNNING`	Task is being processed
`SUCCESS`	Task completed successfully (terminal)
`FAILURE`	Task failed with an error
`RETRYING`	Task is being retried (transitions back to QUEUED)
`CANCELLED`	Task was cancelled (terminal)

Terminal states (SUCCESS, CANCELLED) are immutable — no further transitions allowed.

Task Retries¶

Tasks can be configured with automatic retry policies:

@task
def flaky_operation() -> str:
    # This task will be retried up to 3 times on failure
    ...

# Configure via Task model fields
task = Task(
    name="flaky_operation",
    max_retries=3,
    retry_delay=5.0,       # 5 seconds between retries
    retry_backoff=2.0,     # exponential backoff multiplier
)

Task Timeouts¶

Set per-task timeouts to prevent hung tasks:

task = Task(
    name="long_running",
    timeout=120.0,  # 2 minute timeout
)

Scheduled / Delayed Tasks¶

Submit tasks for future execution using ETA or countdown:

from datetime import datetime, timedelta

# Delay by 30 seconds
task_id = await client.delay("my_task", countdown=30)

# Execute at a specific time
eta = datetime.now() + timedelta(hours=1)
task_id = await client.delay("my_task", eta=eta)

Task Cancellation¶

Cancel queued or running tasks:

CLIPythonGUI

fastworker cancel --task-id <task-id>

cancelled = await client.cancel_task(task_id)

Use the cancel button in the management GUI tasks table.

Task Result¶

When a task completes, you receive a TaskResult:

result = await client.get_task_result(task_id)

if result:
    print(f"Task ID: {result.task_id}")
    print(f"Status: {result.status}")
    print(f"Result: {result.result}")
    print(f"Error: {result.error}")
    print(f"Started: {result.started_at}")
    print(f"Completed: {result.completed_at}")

Best Practices¶

1. Keep Tasks Focused¶

# Good - single responsibility
@task
def send_email(to: str, subject: str, body: str) -> bool:
    return email_service.send(to, subject, body)

# Bad - too many responsibilities
@task
def do_everything(user_id: int):
    user = get_user(user_id)
    send_email(user.email, ...)
    update_database(...)
    notify_slack(...)

2. Handle Errors Gracefully¶

@task
def risky_task(data: dict) -> dict:
    try:
        return process(data)
    except ValueError as e:
        logger.error(f"Validation error: {e}")
        raise  # Re-raise to mark task as failed

3. Use Type Hints¶

@task
def typed_task(name: str, count: int) -> dict[str, any]:
    return {"name": name, "count": count}

4. Avoid Large Arguments¶

# Bad - large data in arguments
@task
def process_file(file_contents: bytes):  # 100MB file!
    return analyze(file_contents)

# Good - pass reference, load in worker
@task
def process_file(file_path: str):
    with open(file_path, 'rb') as f:
        return analyze(f.read())

5. Make Tasks Idempotent When Possible¶

@task
def update_user_status(user_id: int, status: str) -> bool:
    # Safe to call multiple times
    user = get_user(user_id)
    if user.status != status:
        user.status = status
        user.save()
    return True

Serialization¶

JSON (Default)¶

JSON-serializable types:

str, int, float, bool, None
list, dict
Objects with .dict() or .model_dump() methods (Pydantic)

@task
def json_task(data: dict) -> dict:
    return {"result": data["value"] * 2}

Pickle¶

For complex Python objects, use Pickle serialization:

export FASTWORKER_SERIALIZATION_FORMAT=PICKLE

Warning

Only use Pickle in trusted environments. Never use Pickle with untrusted task data.

Common Serialization Issues¶

# Bad - datetime not JSON serializable
@task
def bad_task():
    return {"timestamp": datetime.now()}

# Good - convert to string
@task
def good_task():
    return {"timestamp": datetime.now().isoformat()}