Tasks - Hyrex

Tasks are the fundamental unit of work in Hyrex. They are Python functions that can be executed asynchronously by workers, with automatic retry handling, monitoring, and durability.

Basic Task Definition

Define a task by decorating a function with @hy.task:

from hyrex import HyrexRegistry
from pydantic import BaseModel

hy = HyrexRegistry()

class ProcessContext(BaseModel):
    file_path: str
    options: dict

@hy.task
def process_file(context: ProcessContext):
    # Task logic
    print(f"Processing {context.file_path}")
    return {"status": "completed", "file": context.file_path}

All task parameters must have type hints. Hyrex enforces this requirement to ensure type safety.

Task Configuration

Configure task behavior with decorator options:

@hy.task(
    queue="processing",          # Target queue
    max_retries=3,              # Retry attempts
    timeout_seconds=300,        # Timeout (5 minutes)
    priority=5,                 # 1-10 (higher = more important)
    retry_backoff=lambda n: n*10 # Backoff strategy
)
def configured_task(data: dict):
    # Process with specific configuration
    return {"processed": True}

Configuration Options

queue: Route tasks to specific worker pools
max_retries: Number of retry attempts on failure (default: 0)
timeout_seconds: Maximum execution time before timeout
priority: Task priority from 1-10 (default: 5)
retry_backoff: Function to calculate retry delay

Sending Tasks

Queue tasks for asynchronous execution:

# Send with default configuration
task = process_file.send(ProcessContext(
    file_path="/data/input.csv",
    options={"format": "csv"}
))

# Override configuration at runtime
urgent_task = process_file.with_config(
    queue="high-priority",
    max_retries=5
).send(context)

# Get task ID
print(f"Task ID: {task.id}")

Task Context

Access metadata about the current task execution:

from hyrex import get_hyrex_context

@hy.task
def task_with_context(data: dict):
    # Get current task context
    context = get_hyrex_context()
    
    if context:
        task_id = context.task_id
        attempt_number = context.attempt_number
        queue_name = context.queue
        
        print(f"Processing task {task_id} (attempt {attempt_number})")
        
        # Access parent task info if this is a sub-task
        if context.parent_id:
            print(f"Spawned by task: {context.parent_id}")
    
    # Your task logic here
    return {"processed_by": str(task_id) if context else "unknown"}

Error Handling

Tasks can raise exceptions to trigger retries:

@hy.task(max_retries=3)
def task_with_retries(data: dict):
    try:
        # Attempt operation
        result = external_api_call(data)
        return result
    except TemporaryError:
        # This will trigger a retry
        raise
    except PermanentError as e:
        # This will not retry
        return {"error": str(e), "status": "failed"}

Error Callbacks

Use on_error to handle errors without affecting retry behavior:

def error_handler(e: Exception):
    # Log error, send alert, etc.
    print(f"Task error: {type(e).__name__}: {str(e)}")

@hy.task(
    max_retries=3,
    on_error=error_handler
)
def monitored_task(data: dict):
    # Task logic that might fail
    result = risky_operation(data)
    return result

Retry Behavior

Raising any exception triggers a retry (if retries remain)
Return a value to complete the task (even with errors)
Use retry_backoff for exponential or custom backoff
on_error callbacks run on each failure but don’t affect retries

Task Results

Track task execution:

# Send a task
task = process_file.send(context)

# Task ID is immediately available
print(f"Task ID: {task.id}")

# Get task name
print(f"Task name: {task.task_name}")

# Refresh task status from database
task.refresh()

# Access task runs (attempts)
for run in task.task_runs:
    print(f"Attempt {run.attempt_number}: {run.status}")
    if run.status == "completed":
        print(f"Result: {run.result}")

Best Practices

Use Type-Safe Contexts

class OrderContext(BaseModel):
    order_id: int
    customer_email: str
    items: List[dict]

Set Appropriate Timeouts

@hy.task(timeout_seconds=30)  # Fast operations
@hy.task(timeout_seconds=3600)  # Long-running jobs

Design for Idempotency

@hy.task
def process_payment(context: PaymentContext):
    # Check if already processed
    if payment_exists(context.payment_id):
        return get_payment_result(context.payment_id)
    
    # Process payment
    result = charge_card(context)
    save_payment_result(context.payment_id, result)
    return result

Handle Errors Gracefully

@hy.task(max_retries=3)
def fetch_data(context: FetchContext):
    try:
        return fetch_from_api(context.url)
    except RateLimitError:
        # Retry with backoff
        raise
    except InvalidDataError as e:
        # Don't retry, return error
        return {"error": str(e), "status": "invalid_data"}

Next Steps

Workflows

Orchestrate multiple tasks

Queues

Route tasks to worker pools

Retries

Configure retry strategies

API Reference

Detailed task API

​Basic Task Definition

​Task Configuration

​Configuration Options

​Sending Tasks

​Task Context

​Error Handling

​Error Callbacks

​Retry Behavior

​Task Results

​Best Practices

​Next Steps

Workflows

Queues

Retries

API Reference

Basic Task Definition

Task Configuration

Configuration Options

Sending Tasks

Task Context

Error Handling

Error Callbacks

Retry Behavior

Task Results

Best Practices

Next Steps