celery-manager: See, Control and Pause Your Celery Tasks Without Touching Your Beat Schedule

A dashboard and control API for Celery, for Django, Flask and FastAPI projects.

Celery is great until you need to know what it's actually doing.

A task failed overnight. Which arguments did it get? Which worker ran it? What did it log before the traceback? A customer import has been running for 20 minutes. Is it at row 10 or row 10,000? And one I hit all the time: "Can we skip tomorrow's 9:00 report, just once?" Usually that means editing the beat schedule, deploying, remembering to revert it, and deploying again.

I built celery-manager to answer those questions from one screen. It's a dashboard and control API for Celery that shows what every task is doing, lets you act on tasks, and can pause a single upcoming run of a periodic task without changing your beat schedule or restarting anything.

It works with any Django, Flask or FastAPI project that already uses Celery.

---

What you get

• 🔍 Every run is visible, whether or not the task is decorated: status, arguments, worker, queue, timings, exceptions and full tracebacks.
• 🗂 History and search: paginated history, filters by status, task name or time range, and free-text search over names, IDs and arguments.
• 📈 Live progress: call report_progress() inside a task and watch a progress bar move.
• 📝 Log capture: a task's logging output is shown next to the run.
• 🎛 Control: start any registered task, cancel or terminate runs, and retry them as they were or with edited arguments, one at a time or in bulk.
• ⏸ Schedule control: skip one upcoming periodic run, or pause a whole schedule until you resume it or until a set time.
• 🌳 Workflows: chains, groups and chords drawn as a tree on each run's page.
• 📊 Insights: volume, success rate and runtimes over time, a page for each task, and failures grouped by cause with "retry all".
• 👷 Workers and queues: live worker state, pool controls, runtime rate limits, queue depth and purge.
• 🔗 Share and export: filters live in the URL, plus saved views and CSV or JSON export.
• 🧾 Audit log: every control action is recorded with the user who took it.
• 🎨 Light and dark themes, plus your own wordmark, accent colour and logo.

---

How it works: three small parts

celery-manager has three parts that share one SQLAlchemy-backed store (SQLite by default, or PostgreSQL, MySQL, or any database SQLAlchemy supports):

1. REST API and dashboard. A FastAPI app that serves the JSON API under /api and the dashboard at /. You can mount it in your own app or run it standalone.
2. Events listener (celery-manager events). A long-running process that records status changes from Celery's event bus, for every task, decorated or not.
3. The @track decorator. Optional and per task. It adds log capture, enables pausing and cancelling individual scheduled runs, and records the run as started before your function runs.

Because the listener works from Celery's own events, you get visibility into existing tasks without changing any task code. @track is only for the extra features.

---

Quickstart

1. Install

pip install --pre celery-manager

# optional extras
pip install --pre "celery-manager[standalone]"   # adds uvicorn
pip install --pre "celery-manager[flask]"        # adds Flask and a2wsgi

2. Create the dashboard app

import os

from celery_manager.config import CeleryManagerConfig
from celery_manager.api.app import create_app

config = CeleryManagerConfig(
    celery_app=your_celery_app,
    db_url="sqlite:///celery_manager.db",  # or any SQLAlchemy URL
    secret_key=os.environ["CELERY_MANAGER_SECRET_KEY"],  # private + random
)
app, _ = create_app(config)

Generate a secret key:

python -c "import secrets; print(secrets.token_urlsafe(32))"

create_app refuses to start if the key is empty or still the "change-me" placeholder.

Serve it with any ASGI server:

uvicorn your_module:app

Local development over plain HTTP? The session cookie is Secure by default, so your browser won't send it to http://localhost and you'll look logged out after signing in. Pass cookie_secure=False for local development only.

3. Turn on Celery events

Add these to the Celery config shared by web, worker and beat:

app.conf.worker_send_task_events = True   # same as running the worker with -E
app.conf.task_send_sent_event = True      # publisher emits task-sent (carries the queue)

4. Run the events listener

Run it next to your worker and beat:

celery-manager events --celery-app your_module:your_celery_app --db-url sqlite:///celery_manager.db

Use the same --db-url as in your config.

5. Create your first user

There's no default account, which is intentional. Create one:

from celery_manager.auth.users import create_user
from celery_manager.db.session import init_db, get_session_factory

Session = get_session_factory(init_db(config))
create_user(Session(), "admin", "a-real-password")

Sign in, and every task run your workers process shows up in the dashboard.

---

Going further with @track

Put @track under @app.task, and call celery_manager.setup() in the worker process. create_app() only sets up the dashboard process, so without setup() the decorator does nothing.

# myproject/tasks.py — imported by the Celery worker
import celery_manager
from celery_manager.config import CeleryManagerConfig

from myproject.celery import app

celery_manager.setup(CeleryManagerConfig(
    celery_app=app,
    db_url="sqlite:///celery_manager.db",  # the same DB the dashboard uses
    secret_key="a-real-secret-in-production",
))


@app.task(bind=True)
@celery_manager.track(schedule_name="send-report-every-2-hours")
def send_report(self):
    ...

schedule_name should match the key in your beat_schedule. bind=True is optional, so @track also works on plain tasks:

@app.task
@celery_manager.track()
def send_welcome_email(address):
    ...

Live progress bars

from celery_manager import report_progress


@app.task(bind=True)
@celery_manager.track()
def import_customers(self, row_count: int = 60):
    for row in range(1, row_count + 1):
        handle_row(row)
        report_progress(row, row_count, f"row {row} of {row_count}")
    return f"imported {row_count} rows"

The signature is report_progress(current, total=None, message=None). A few details:

• It's throttled. Database writes happen at most once every 0.5 seconds per run, so you can call it in a tight loop. The final call (current >= total) always gets written.
• It never raises into your task. A progress problem won't crash a real job.
• Leave out total for open-ended work. The dashboard then shows a running count and message instead of a percentage.

Logs and tracebacks next to the run

When a tracked task fails, its page shows the traceback, the captured log output and a Retry button together:

---

The headline feature: pause a schedule without editing it

celery-manager never edits your beat schedule. Beat keeps sending the task as usual, and @track makes that run skip itself when you've paused it. So there's no config change, no deploy, and nothing to remember to revert.

From the Schedules page you can:

• Skip one upcoming run. Open a schedule, see its next runs, and skip or cancel just one. The rest of the schedule carries on.
• Pause the whole schedule for 1 hour, until tomorrow at 9:00, or until a custom time.

When you resume, the schedule continues from the current time. Runs missed while paused are not run all at once afterwards.

---

Full control from the dashboard

• Start task. Pick any registered task. Its arguments are prefilled from the function signature, and you can choose a queue or a delay.
• Cancel a run that hasn't started, or Terminate one that's already running.
• Retry, or Retry with changes to edit the arguments first. You can select several failed runs and retry them together.
• Bulk cancel every waiting or running task that matches your filters. The count is shown before you confirm.
• Workers. Shut a worker down, add or remove pool processes, and start or stop consuming from a queue.
• Rate limits. Change a task's rate limit on running workers without a deploy. The change lasts until the worker restarts.
• Purge a queue. You have to type the queue name to confirm, so it's hard to do by accident.

Everything in the dashboard is also available through the REST API, so you can script it or connect it to your own tools.

---

Workflows, insights and grouped failures

Chains, groups and chords are drawn as a tree, so when one member of a chord fails you can see exactly which one:

Insights show volume, success rate and runtime trends, with a page for each task:

Failures are grouped by exception signature. Instead of 300 separate failed rows, you see "these 300 runs failed with the same ConnectionError", and you can retry the whole group at once:

---

Plugging it into your stack

The dashboard is a standard ASGI app, so it fits in most places.

FastAPI. Mount it next to your own routes:

from fastapi import FastAPI

from celery_manager.api.app import create_app
from celery_manager.config import CeleryManagerConfig

dashboard, _session_factory = create_app(CeleryManagerConfig(
    celery_app=celery_app,
    db_url="sqlite:///celery_manager.db",
    secret_key=SECRET,
))

app = FastAPI()
app.mount("/celery-manager", dashboard)

Flask. Wrap it with a2wsgi (install celery-manager[flask]):

from a2wsgi import ASGIMiddleware
from flask import Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware

from celery_manager.api.app import create_app

app = Flask(__name__)
dashboard, _ = create_app(config)

app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {
    "/celery-manager": ASGIMiddleware(dashboard),
})

Django. The recommended setup is to run the dashboard as its own uvicorn process next to Django, optionally behind your reverse proxy:

uvicorn yourmodule:dashboard --port 8001 --root-path /celery-manager

The repository includes a working reference in examples/django_demo/run_dashboard.py.

For live push updates, run the dashboard as its own process rather than through the Flask/WSGI adapter.

---

Production notes

• Use a real database. SQLite is fine for local development. In production, point db_url at PostgreSQL or MySQL.
• Run migrations as a deploy step with celery-manager migrate, and set auto_migrate=False so the app doesn't upgrade the schema itself at startup:

  celery-manager migrate --db-url postgresql://admin@db/celery_manager

• Keep cookie_secure=True (the default) anywhere outside local development.
• macOS tip: with Celery 5.6 and billiard 4.3, prefork workers on macOS can fail every task with ValueError: not enough values to unpack. This is an upstream issue. Use --pool=threads or --pool=solo locally.

---

Status: early alpha

The current release is 0.1.0a4 and needs Python 3.10+. To be upfront about what that means:

• The API may still change.
• There's no role system yet. Any signed-in user can perform any action. Every action is recorded in the audit log, but you should still only give access to people you trust.

My recommendation: use it on internal tools for now, and pin the exact version.

---

Links

• ⭐ GitHub: github.com/Pythondeveloper6/celery-manager
• 📦 PyPI: pypi.org/project/celery-manager
• 📖 Docs: pythondeveloper6.github.io/celery-manager

If you run Celery in production, I'd really like to know what question you most often can't answer about your tasks. That's what decides what I build next. Issues, ideas and stars on GitHub are all welcome.

Thanks for reading!

Mahmoud Ahmed