django-tasks-repid

A django-tasks backend powered by repid.

Requirements

• Python 3.12+
• Django 6.0+
• django-tasks 0.12+
• repid 1.6+

Installation

pip install django-tasks-repid

Configuration

Add the backend to your TASKS setting:

TASKS = {
    "default": {
        "BACKEND": "django_tasks_repid.backend.RepidBackend",
        "QUEUES": ["default"],
        "OPTIONS": {
            "BROKER": "redis://localhost:6379",
            "RESULT_BACKEND": "redis://localhost:6379",
        },
    }
}

Options

Supported brokers

Defining tasks

from django_tasks import task

@task()
def send_welcome_email(user_id: int) -> None:
    ...

@task()
async def generate_report(report_id: int) -> dict:
    ...

Enqueueing tasks

# Sync
result = send_welcome_email.enqueue(user_id=42)

# Async
result = await send_welcome_email.aenqueue(user_id=42)

# Deferred
from datetime import datetime, timedelta, timezone

result = send_welcome_email.using(
    run_after=datetime.now(tz=timezone.utc) + timedelta(hours=1)
).enqueue(user_id=42)

Retrieving results

get_result() and aget_result() require RESULT_BACKEND to be configured.

from django_tasks import default_task_backend

result = send_welcome_email.enqueue(user_id=42)

# Later — fetch updated status
result = default_task_backend.get_result(result.id)

if result.is_finished:
    print(result.return_value)

Running a worker

Use repid's worker to consume and execute tasks:

# worker.py
import django
django.setup()

import asyncio
from repid import Connection, Queue, Repid, Router
from repid.connections import RedisMessageBroker

from myapp.tasks import send_welcome_email

router = Router()
router.include_func(send_welcome_email.func)

async def main():
    async with Repid(Connection(RedisMessageBroker("redis://localhost:6379"))).magic():
        await Queue("default").declare()
        worker = router.create_worker(Queue("default"))
        await worker.run()

asyncio.run(main())

Development / testing

The in-memory broker (no BROKER configured) is useful for tests:

# settings.py
TASKS = {
    "default": {
        "BACKEND": "django_tasks_repid.backend.RepidBackend",
        "OPTIONS": {},
    }
}

Run the test suite:

uv run pytest

License

MIT