Background Task Queuing¶

Ushka includes a background task queuing system that allows you to offload long-running or resource-intensive operations from the main web request-response cycle. This is ideal for tasks like sending emails, processing images, or calling slow external APIs, ensuring your web responses remain fast.

Enabling and Configuring the Queue¶

The queuing system is disabled by default. To enable it, add a [queue] section to your ushka.toml file.

# in ushka.toml

[queue]
enabled = true
backend = "in_memory"

Backends¶

Ushka supports two queuing backends out of the box:

in_memory (default): This is a simple, process-local in-memory queue. It's useful for development and testing, but tasks will be lost if the application restarts. It is not suitable for production.
redis: This uses a Redis server as a robust and persistent message broker. This is the recommended choice for production.

To use the redis backend, you must also provide a redis_url:

[queue]
enabled = true
backend = "redis"
redis_url = "redis://localhost:6379/0"

You will also need to install the redis library:

pip install redis

Creating a Task¶

A "task" is simply a Python function. For the queue worker to find it, this function must be defined in an importable module within your project. A good practice is to create a tasks.py file in one of your apps.

Important: The function must be self-contained or use Ushka's dependency injection to get its dependencies, as it will be run in a separate worker process.

# in apps/home/tasks.py
import time

def send_welcome_email(user_email: str):
    print(f"Sending welcome email to {user_email}...")
    # Simulate a slow network operation
    time.sleep(10)
    print(f"Email sent to {user_email}!")

Dispatching a Task¶

To add a task to the queue, you need to use the QueueClient service. You can inject this service into your views or other services.

# in apps/home/views.py
from ushka.http import Request
from ushka.contrib.queuing.queue import QueueClient
from . import tasks # Import your tasks module

async def post_register(request: Request, queue: QueueClient):
    form = await request.form()
    user_email = form.get("email")

    # ... logic to create the user ...

    # Now, dispatch the email task to the background queue
    await queue.dispatch(tasks.send_welcome_email, user_email)

    return {"status": "User registered! Welcome email is on its way."}

The .dispatch() method takes the task function itself, followed by any positional or keyword arguments that the function requires.

Processing Tasks¶

Dispatching a task only adds it to the queue. A separate "worker" process is needed to pull tasks from the queue and execute them.

Ushka provides a CLI command to run a worker.

# Start a worker that processes one task from the queue and then exits
ushka queue process

# Start a worker that processes up to 10 tasks before exiting
ushka queue process --max-tasks 10

In a production environment, you would typically run this command on a schedule (e.g., as a cron job every minute) or have it run continuously using a process manager like Supervisor.