Agent Skills: TasksModule - Background Jobs

TasksModule - Background Jobs

TasksModule provides SQL-based async task processing with DI injection and automatic retries.

Quick Start

from myfy.core import Application
from myfy.data import DataModule
from myfy.tasks import TasksModule, task

app = Application()
app.add_module(DataModule())
app.add_module(TasksModule(auto_create_tables=True))

# Define a task
@task
async def send_email(to: str, subject: str, body: str) -> None:
    await email_service.send(to, subject, body)

# Dispatch from a route
@route.post("/notifications")
async def notify_user(body: NotifyRequest) -> dict:
    task_id = await send_email.send(
        to=body.email,
        subject="Welcome!",
        body="Thanks for signing up.",
    )
    return {"task_id": task_id}

Configuration

Environment variables use the MYFY_TASKS_ prefix:

| Variable | Default | Description | |----------|---------|-------------| | MYFY_TASKS_DEFAULT_MAX_RETRIES | 3 | Default retry attempts | | MYFY_TASKS_RETRY_DELAY_SECONDS | 60.0 | Seconds between retries | | MYFY_TASKS_WORKER_CONCURRENCY | 4 | Concurrent tasks per worker | | MYFY_TASKS_POLL_INTERVAL | 1.0 | Seconds between queue polls | | MYFY_TASKS_TASK_TIMEOUT | 300.0 | Max seconds per task |

Defining Tasks

Basic Task

from myfy.tasks import task

@task
async def process_order(order_id: int) -> str:
    # Process the order
    return f"Processed order {order_id}"

Task with DI Injection

Services are automatically injected at runtime:

from myfy.tasks import task
from myfy.data import AsyncSession

@task
async def sync_user_data(user_id: int, session: AsyncSession) -> None:
    # session is TASK-scoped (injected per task execution)
    user = await session.get(User, user_id)
    await sync_to_external_service(user)

Task with Custom Options

@task(max_retries=5, retry_on=[ConnectionError, TimeoutError])
async def upload_file(file_path: str) -> str:
    # Retries up to 5 times on connection/timeout errors
    return await s3.upload(file_path)

Dispatching Tasks

Basic Dispatch

# Returns immediately with task_id
task_id = await send_email.send(to="user@example.com", subject="Hi")

Dispatch Options

task_id = await send_email.send(
    to="user@example.com",
    subject="Hi",
    _priority=10,      # Higher priority = executes first
    _delay=60,         # Wait 60 seconds before executing
    _max_retries=5,    # Override default retries
)

Getting Results

result = await send_email.get_result(task_id, timeout=60)

if result.is_completed:
    print(f"Success: {result.value}")
elif result.is_failed:
    print(f"Error: {result.error}")
elif result.is_pending:
    print("Still processing...")

TaskContext for Progress

Report progress from long-running tasks:

from myfy.tasks import task, TaskContext

@task
async def import_users(file_path: str, ctx: TaskContext) -> int:
    users = load_users_from_file(file_path)
    total = len(users)

    for i, user in enumerate(users):
        await create_user(user)
        await ctx.update_progress(
            current=i + 1,
            total=total,
            message=f"Importing user {i + 1}/{total}",
        )

    return total

Check progress from caller:

result = await import_users.get_result(task_id)
if result.progress:
    current, total = result.progress
    print(f"Progress: {current}/{total} - {result.progress_message}")

Running Workers

Start a worker process:

myfy tasks worker

With options:

myfy tasks worker --concurrency 8 --poll-interval 0.5

Workers:

• Poll the database for pending tasks
• Execute tasks with full DI injection
• Handle retries automatically
• Report progress and results
• Gracefully shutdown on SIGTERM

Task States

| Status | Description | |--------|-------------| | pending | Queued, waiting for worker | | running | Being executed by worker | | completed | Finished successfully | | failed | Failed after all retries | | cancelled | Manually cancelled |

Error Handling

Tasks automatically retry on failure:

@task(max_retries=3, retry_on=[APIError])
async def call_api(url: str) -> dict:
    response = await http.get(url)
    if response.status >= 500:
        raise APIError("Server error")  # Will retry
    return response.json()

After all retries fail:

• Task status becomes failed
• Error message and traceback are stored
• Can be retrieved via get_result()

Parameter Classification

| Type | Behavior | |------|----------| | Primitives (str, int, float, bool) | Serialized as task args | | Lists, dicts | Serialized as task args | | TaskContext | Injected by worker | | Services (other types) | DI injected at runtime |

@task
async def complex_task(
    order_id: int,           # Serialized (primitive)
    items: list[str],        # Serialized (list)
    ctx: TaskContext,        # Injected (context)
    session: AsyncSession,   # DI injected (service)
    settings: AppSettings,   # DI injected (service)
) -> None:
    ...

Best Practices

1. Keep tasks idempotent - Safe to retry on failure
2. Serialize only primitives - Complex objects should be loaded in task
3. Use TaskContext - Report progress for long tasks
4. Set appropriate timeouts - Prevent zombie tasks
5. Monitor worker logs - Watch for repeated failures
6. Use priorities - Critical tasks get processed first
7. Handle cleanup - TaskContext supports cleanup callbacks