Tasks Module

TasksModule - Background Jobs

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

Quick Start

pythonCopy code
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:

Defining Tasks

Basic Task

pythonCopy code
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:

pythonCopy code
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

pythonCopy code
@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

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

Dispatch Options

pythonCopy code
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

pythonCopy code
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:

pythonCopy code
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:

pythonCopy code
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:

bashCopy code
myfy tasks worker

With options:

bashCopy code
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

Error Handling

Tasks automatically retry on failure:

pythonCopy code
@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

pythonCopy code
@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