Django Integration

Admin views, settings, autodiscovery, management commands — no Django ORM models required.

flexiq provides Django admin views for browsing jobs, inspecting dead letters, and viewing queue statistics — all without Django ORM models.

Installation

bash

pip install flexiq[django]

Setup

Add the integration to INSTALLED_APPS. This is required for the admin templates to load — Django's app-directories template loader only scans installed apps — and for task autodiscovery:

python

# settings.py
INSTALLED_APPS = [
    # ...
    "django.contrib.admin",
    "flexiq.contrib.django",
]

Template loading

Keep the default APP_DIRS = True in your TEMPLATES setting (Django's default) so the bundled admin templates are discovered.

Option 1: register on the default admin site

In your project's admin.py or urls.py:

python

from flexiq.contrib.django.admin import register_flexiq_admin

register_flexiq_admin()

This adds flexiq views to the default admin.site.

Option 2: custom admin site

Use FlexiQAdminSite for a dedicated admin:

python

from flexiq.contrib.django.admin import FlexiQAdminSite

admin_site = FlexiQAdminSite(name="flexiq_admin")

Django settings

The following settings can be defined in your Django settings.py:

Setting	Default	Description
`FLEXIQ_AUTODISCOVER_MODULE`	`"tasks"`	Module name auto-discovered in each installed app on startup.
`FLEXIQ_ADMIN_PER_PAGE`	`50`	Rows per page in the admin jobs and dead letters views.
`FLEXIQ_ADMIN_TITLE`	`"FlexiQ"`	Browser tab title for `FlexiQAdminSite`.
`FLEXIQ_ADMIN_HEADER`	`"FlexiQ Admin"`	Site header shown in `FlexiQAdminSite`.
`FLEXIQ_WATCH_INTERVAL`	`2`	Polling interval in seconds for `manage.py flexiq_info --watch`.
`FLEXIQ_DASHBOARD_HOST`	`"127.0.0.1"`	Default bind host for `manage.py flexiq_dashboard`.
`FLEXIQ_DASHBOARD_PORT`	`8080`	Default bind port for `manage.py flexiq_dashboard`.
`FLEXIQ_DASHBOARD_AUTH`	`False`	Enable dashboard session authentication by default (same as passing `--auth` to `manage.py flexiq_dashboard`).

Example:

python

# settings.py
FLEXIQ_AUTODISCOVER_MODULE = "jobs"   # import myapp.jobs instead of myapp.tasks
FLEXIQ_ADMIN_PER_PAGE = 25
FLEXIQ_ADMIN_TITLE = "MyApp Tasks"
FLEXIQ_ADMIN_HEADER = "MyApp Task Queue"
FLEXIQ_DASHBOARD_HOST = "0.0.0.0"
FLEXIQ_DASHBOARD_PORT = 9000

Queue configuration

Create a flexiq queue instance in your Django project — this is the object your tasks are decorated on and the one the worker CLI runs against (--app myproject.tasks:queue).

python

# myproject/tasks.py
from flexiq import Queue

queue = Queue(db_path="flexiq.db")

@queue.task()
def send_welcome_email(user_id: int):
    from myapp.models import User
    user = User.objects.get(id=user_id)
    user.email_user("Welcome!", "Thanks for signing up.")

Task modules are located via Django's standard app autodiscovery: flexiq.contrib.django's AppConfig.ready() imports a tasks module (the name is configurable via FLEXIQ_AUTODISCOVER_MODULE) from every installed app, which registers any @queue.task()-decorated functions it imports.

get_queue() is a separate Queue instance

flexiq.contrib.django.settings.get_queue() — used internally by the admin views and the manage.py flexiq_* commands — builds its own Queue from the FLEXIQ_* Django settings. It is not the same Python object as the queue you create in myproject/tasks.py. Point both at the same FLEXIQ_DB_PATH / FLEXIQ_BACKEND / FLEXIQ_DB_URL so they read the same underlying storage — otherwise the admin views show an empty or different queue than the one your worker is processing.

Lazy imports

Import Django models inside the task function body to avoid app registry issues during startup.

Admin views

The integration provides the following views under /admin/flexiq/:

Dashboard — Queue statistics overview (pending, running, completed, failed, dead, cancelled)
Jobs — Paginated job list with status, queue, and task name filters
Job Detail — Full job payload, error history, retry count, and metadata
Dead Letters — Browse and retry dead letter entries

Running the worker

bash

DJANGO_SETTINGS_MODULE=myproject.settings flexiq worker --app myproject.tasks:queue

Full example

Project structure

code

myproject/
  settings.py
  urls.py
  tasks.py          # Queue + task definitions
  myapp/
    admin.py         # Register flexiq admin views
    views.py
    models.py

`tasks.py`

python

from flexiq import Queue

queue = Queue(db_path="flexiq.db")

@queue.task(max_retries=3)
def send_welcome_email(user_id: int):
    from myapp.models import User
    user = User.objects.get(id=user_id)
    user.email_user("Welcome!", "Thanks for signing up.")

@queue.task(rate_limit="60/h")
def generate_monthly_report(month: int, year: int):
    from myapp.reports import build_report
    return build_report(month, year)

`myapp/admin.py`

python

from django.contrib import admin
from flexiq.contrib.django.admin import register_flexiq_admin

register_flexiq_admin()

`myapp/views.py`

python

from django.http import JsonResponse
from myproject.tasks import send_welcome_email

def signup_view(request):
    user = create_user(request.POST)
    send_welcome_email.delay(user.id)
    return JsonResponse({"status": "ok"})