Celery Task Queue Architecture for Background Jobs in the Plane API

The Plane API leverages Celery with RabbitMQ as the message broker, Django ORM as the result backend, and a modular plane/bgtasks/ package to execute asynchronous background jobs via @shared_task decorators.

The open-source project management platform Plane uses Celery to handle resource-intensive operations outside the request-response cycle. Understanding the Celery task queue architecture for background jobs is essential for developers extending the API or deploying self-hosted instances. The implementation centralized in plane/celery.py and plane/settings/common.py provides a scalable, Django-integrated system for task processing and periodic scheduling.

Core Architecture Components

Broker Configuration and Django Settings

The message broker connection is configured in plane/settings/common.py using environment variables or an explicit AMQP_URL. The system defaults to RabbitMQ but can be overridden for cloud-hosted AMQP services.

# plane/settings/common.py

if AMQP_URL:
    CELERY_BROKER_URL = AMQP_URL
else:
    CELERY_BROKER_URL = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASSWORD}@{RABBITMQ_HOST}:{RABBITMQ_PORT}/{RABBITMQ_VHOST}"

The result backend defaults to the Django ORM, storing task execution states and return values in the database. This eliminates the need for additional Redis or cache infrastructure in standard deployments.

Celery App Initialization

The central Celery application is instantiated in plane/celery.py as a standalone module. This file creates the app instance, loads Django configuration using the CELERY namespace, and registers the beat schedule for periodic tasks.

# plane/celery.py

app = Celery("plane")
app.config_from_object("django.conf:settings", namespace="CELERY")

This configuration approach ensures that all Celery-specific settings (broker URL, serialization formats, and import paths) are configurable through standard Django settings files.

Task Discovery via CELERY_IMPORTS

Unlike auto-discovery, Plane uses explicit task registration through the CELERY_IMPORTS tuple in plane/settings/common.py. This guarantees that worker processes load all background task modules at startup, preventing race conditions in module loading.

# plane/settings/common.py

CELERY_IMPORTS = (
    "plane.bgtasks.issue_automation_task",
    "plane.bgtasks.exporter_expired_task",
    "plane.bgtasks.file_asset_task",
    # ... additional task modules

)

Implementing Background Tasks

Defining Tasks with @shared_task

Individual background jobs reside in separate modules under plane/bgtasks/. Each task is decorated with @shared_task, making it importable without requiring a direct reference to the Celery app instance. This pattern is critical for avoiding circular import issues in Django projects.

# plane/bgtasks/email_notification_task.py

from celery import shared_task

@shared_task
def stack_email_notification():
    # Logic to batch-send pending email notifications

    pass

When creating custom tasks, import Django models lazily inside the function to prevent circular dependencies:

# plane/bgtasks/my_custom_task.py

from celery import shared_task

@shared_task
def send_welcome_email(user_id):
    from plane.users.models import User  # Lazy import

    user = User.objects.get(pk=user_id)
    # Email sending logic

Periodic Scheduling with Celery Beat

The app.conf.beat_schedule dictionary in plane/celery.py defines periodic tasks using crontab or timedelta schedules. The beat scheduler runs as a separate process and triggers tasks according to these definitions.

# plane/celery.py

from celery.schedules import crontab, schedule
from datetime import timedelta

app.conf.beat_schedule = {
    "check-every-five-minutes-to-send-email-notifications": {
        "task": "plane.bgtasks.email_notification_task.stack_email_notification",
        "schedule": crontab(minute="*/5"),
    },
    "push-instance-metrics": {
        "task": "plane.license.bgtasks.telemetry_metrics.push_instance_metrics",
        "schedule": schedule(run_every=timedelta(minutes=METRICS_PUSH_INTERVAL_MINUTES)),
    },
}

Enqueuing Tasks from API Views

Trigger tasks asynchronously using the .delay() method or .apply_async() for advanced options like countdowns or specific queues:

# Anywhere in the API code

from plane.bgtasks.my_custom_task import send_welcome_email

def register_user(request):
    # ... user creation logic ...

    send_welcome_email.delay(new_user.id)  # Enqueues immediately

    return JsonResponse({"status": "queued"})

Production Deployment and Worker Management

The architecture requires two distinct process types in production:

1. Worker processes: Consume and execute tasks from the broker
2. Beat scheduler: Triggers periodic tasks based on the schedule

Run these using the Plane Django project name:

# Worker process

celery -A plane worker -l info

# Beat scheduler (periodic tasks)

celery -A plane beat -l info

Both processes share the same Django settings and can be scaled horizontally by deploying additional worker containers behind a load balancer. The Django ORM result backend persists task outcomes, allowing API endpoints to query job completion status via AsyncResult objects.

Key Files and Modules

File	Purpose
plane/celery.py	Central Celery app definition, beat schedule configuration, and logging setup
plane/settings/common.py	Broker URL construction, CELERY_IMPORTS tuple, and serializer settings
plane/bgtasks/email_notification_task.py	Example task implementing batched email notifications
plane/bgtasks/issue_automation_task.py	Automated maintenance tasks for issue lifecycle management
plane/license/bgtasks/telemetry_metrics.py	License-specific background tasks for instance telemetry

Summary

• Broker: RabbitMQ (configured via AMQP_URL or environment variables) handles message queuing between the API and workers.
• App Structure: A single Celery app in plane/celery.py loads Django settings and registers the beat schedule.
• Task Discovery: Explicit CELERY_IMPORTS in settings ensures all modules under plane/bgtasks/ are available to workers.
• Task Definition: Use @shared_task decorators to create importable, self-contained background functions.
• Scheduling: Celery Beat triggers periodic jobs defined in app.conf.beat_schedule with crontab or timedelta intervals.
• Execution: Workers consume tasks from the broker and store results in the Django ORM backend.

Frequently Asked Questions

What message broker does the Plane API use for Celery?

The Plane API uses RabbitMQ as the primary message broker, configured through environment variables (RABBITMQ_USER, RABBITMQ_PASSWORD, RABBITMQ_HOST) or a single AMQP_URL string in plane/settings/common.py. This broker stores task messages until worker processes consume them.

How does the Plane API discover and load background task modules?

The API uses explicit task discovery via the CELERY_IMPORTS tuple in plane/settings/common.py rather than auto-discovery. This tuple lists all Python modules containing @shared_task decorators, ensuring workers load these modules at startup before accepting any jobs.

How do I add a new periodic background task in Plane?

Add your @shared_task function to a module under plane/bgtasks/, include the module path in CELERY_IMPORTS, and register a schedule entry in plane/celery.py under app.conf.beat_schedule. Specify the fully-qualified task path and a schedule using crontab() or schedule() from celery.schedules.

What is the difference between @shared_task and @app.task in the Plane codebase?

The Plane API exclusively uses @shared_task in plane/bgtasks/ modules. This decorator does not require a direct app instance reference, preventing circular import issues common in Django projects. While @app.task (using the specific Celery app instance) is available, @shared_task provides better modularity for the Plane architecture.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}