Conductor

A lightweight, production-ready async task queue for Python teams that don't need Redis.

Conductor orchestrates reliable, distributed task execution with exactly-once semantics, on the PostgreSQL, MySQL/MariaDB or SQLite database you already have. Simple API, observable by default, deploy anywhere.

Documentation: Installation · Configuration · API Reference · Deployment · Troubleshooting · docs index

---

• Why Conductor?
• Quick Start
• Core Features
• Installation
• Usage Examples
• Configuration
• Deployment
• Architecture
• API Reference
• Best Practices
• Troubleshooting
• Comparison to Alternatives
• Roadmap
• Contributing
• License

---

Table of Contents

• Why Conductor?
• Quick Start
• Core Features
• Installation
• Usage Examples
• Configuration
• Deployment
• Architecture
• API Reference
• Best Practices
• Troubleshooting
• Comparison to Alternatives
• Roadmap
• Local Development
• Contributing
• License

---

Why Conductor?

The Problem

Every Python team that outgrows Celery or doesn't want Redis dependency ends up rebuilding task queue logic:

• Celery is overkill: Over-engineered for most teams, requires Redis/RabbitMQ, steep learning curve
• RQ is too simple: Works for basic cases, but lacks reliability patterns (exactly-once semantics, dead letter queues, observability)
• dramatiq is okay: Still requires external brokers, doesn't solve the "I just need PostgreSQL" case
• Custom solutions break: Teams implement retry logic, task deduplication, and worker pools from scratch—each time

The Solution

Conductor solves this by:

✅ No external dependencies – PostgreSQL only. Deploy to any server, container, or serverless environment
✅ Exactly-once semantics – Tasks execute once, guaranteed. Built-in idempotency and deduplication
✅ Observable from day one – Structured logging with task-level context, Prometheus metrics (GET /metrics) and health checks (GET /health) ✅ Production-ready – Exponential backoff, circuit breakers, graceful shutdown, dead letter queues
✅ Simple API – Submit a task in one line. Register a worker in two
✅ Built for async – Native asyncio support. No threads, no blocking calls

---

Quick Start

Installation

pip install conductor-task-queue

Prerequisites:

• Python 3.11+
• PostgreSQL 12+ — or MySQL 8.0+ / MariaDB 10.6+ (pip install "conductor-task-queue[mysql]"), or SQLite for embedded single-process use (pip install "conductor-task-queue[sqlite]", no server required)

Basic Example

1. Submit a task (app.py):

import asyncio
from conductor import TaskQueue, RetryPolicy


async def main() -> None:
    async with TaskQueue(
        database_url="postgresql://user:password@localhost/conductor"
    ) as queue:
        task_id = await queue.submit(
            task_type="send_email",
            payload={
                "to": "user@example.com",
                "subject": "Hello",
                "body": "Welcome to Conductor!",
            },
            retry_policy=RetryPolicy(
                max_retries=3,
                backoff_strategy="exponential",
            ),
        )
        print(f"Task submitted: {task_id}")


asyncio.run(main())

2. Process tasks with a worker (worker.py):

import asyncio
from conductor import Worker


async def main() -> None:
    async with Worker(
        database_url="postgresql://user:password@localhost/conductor"
    ) as worker:

        @worker.task("send_email")
        async def send_email(payload: dict) -> dict:
            to = payload["to"]
            subject = payload["subject"]
            body = payload["body"]

            # Send email logic here
            print(f"Sending email to {to}: {subject}")

            # Simulate sending
            await asyncio.sleep(0.5)

            return {"status": "sent"}

        await worker.run()


asyncio.run(main())

3. Run it:

# Terminal 1: Start the worker
python worker.py

# Terminal 2: Submit a task
python app.py

That's it. Your task is submitted, retried on failure, and executed exactly once.

---

Core Features

1. Exactly-Once Semantics

Tasks execute exactly once, even if workers crash mid-execution:

@worker.task("process_payment")
async def process_payment(payload):
    payment_id = payload["id"]
    
    # If a worker crashes here, the task will NOT be retried
    # (it was already marked as completed)
    
    if await is_payment_processed(payment_id):
        return {"status": "already_processed"}
    
    await charge_card(payload)
    await mark_payment_processed(payment_id)
    
    return {"status": "success"}

2. Retry Logic with Exponential Backoff

Configurable retry policies with automatic backoff:

from conductor import RetryPolicy

await queue.submit(
    task_type="flaky_api_call",
    payload={"url": "https://api.example.com/data"},
    retry_policy=RetryPolicy(
        max_retries=5,
        backoff_strategy="exponential",
        initial_delay=1.0,  # seconds
        max_delay=300.0,
    ),
)

Retry attempts (with default exponential backoff, max_retries=5):

• Attempt 1: Immediate
• Attempt 2: ~1 second delay
• Attempt 3: ~2 second delay
• Attempt 4: ~4 second delay
• Attempt 5: ~8 second delay
• Attempt 6: Moved to dead letter queue

3. Dead Letter Queue

Tasks that fail all retries go to a dead letter queue for manual inspection:

from conductor import DeadLetterQueue

dlq = DeadLetterQueue(database_url="postgresql://...")

# Inspect failed tasks
failed_tasks = dlq.list_tasks(limit=10)

for task in failed_tasks:
    print(f"Task {task.task_id} failed: {task.error_message}")
    print(f"Attempts: {task.attempts}")

# Retry a failed task manually
await dlq.retry_task(task_id="abc-123")

# Or mark as permanently failed
await dlq.discard_task(task_id="abc-123", reason="Known issue, will retry manually later")

4. Built-in Observability

Conductor logs every task transition with structured context using Python's standard logging module. All log messages include task ID, task type, worker ID, and duration where applicable.

import logging

logger = logging.getLogger("conductor.core.worker")

# Automatically logged events:
# - Task submitted  (INFO)
# - Task started    (DEBUG)
# - Task completed  (INFO)  — includes duration_ms
# - Task failed     (ERROR) — includes error_message
# - Task retrying   (WARNING)
# - Task moved to DLQ (WARNING)

Structured JSON logging, Prometheus metrics (GET /metrics), and health checks (GET /health) are included and enabled by default. The worker exposes them on METRICS_PORT (default 8000).

5. Graceful Shutdown

Workers shut down cleanly, completing in-flight tasks. Use the run() method which handles SIGTERM and SIGINT automatically:

import asyncio
from conductor import Worker


async def main() -> None:
    async with Worker(database_url="postgresql://...") as worker:
        await worker.run()  # runs until SIGTERM/SIGINT


asyncio.run(main())

To shut down programmatically, call await worker.shutdown(). All in-flight tasks complete before the worker disconnects.

### 6. Task Routing & Priority Queues

Submit tasks to named ``route``\ s and run dedicated worker pools, or
prioritize work so higher-priority tasks run first.

```python
# Route + priority
await queue.submit(
    "send_sms",
    {"to": "+1234567890"},
    route="critical",  # only workers subscribing to "critical" poll this
    priority=50,        # higher runs first (range -100..100)
)

# Route-specific worker
worker = Worker(database_url="...", routes=["critical"])
# Or poll everything:
worker = Worker(database_url="...", routes=None)  # None = all routes

Priority is honored by the polling query (ORDER BY priority DESC, created_at ASC), so higher-priority tasks are dispatched first even when submitted later. Routed/prioritized tasks keep their route/priority when recovered from the dead-letter queue.

7. Scheduled & Recurring Tasks

Defer one-off work with scheduled_for, or register cron-driven recurring definitions that the RecurringScheduler fires automatically.

# One-off: run at most 1 hour from now
await queue.submit("cleanup", {"keep": 30},
                  scheduled_for=datetime.now(timezone.utc) + timedelta(hours=1))

# Recurring: create one task instance per cron fire (UTC)
rid = await queue.schedule_recurring("cleanup", {"keep": 30}, "0 2 * * *")

# Run the scheduler (standalone, or embed in a worker)
from conductor import RecurringScheduler
async with RecurringScheduler(database_url="...") as sched:
    await sched.run()

Definitions are claimed with FOR UPDATE SKIP LOCKED (safe for multiple schedulers); next_run_at advances to the next cron time (UTC, skipping missed runs). Manage them via list_recurring_tasks(), pause_recurring(), resume_recurring(), and delete_recurring_task().

8. gRPC API (polyglot workers)

A Worker can embed an async gRPC server so non-Python clients can execute tasks and inspect worker state.

worker = Worker(database_url="...", grpc_enabled=True, grpc_port=50051)

service ConductorWorker {
  rpc ProcessTask(TaskRequest) returns (TaskResponse);
  rpc RegisterHandler(RegisterRequest) returns (RegisterResponse);
  rpc GetWorkerStatus(StatusRequest) returns (WorkerStatus);
}

• ProcessTask executes a task through the worker's registered handler (persist=true also records the outcome in conductor_tasks).
• RegisterHandler declares a handler for a task_type (idempotent).
• GetWorkerStatus returns worker health and statistics.

Generated stubs are committed (conductor/grpc/) so no protoc is needed at install; regenerate with python scripts/generate_grpc.py. See examples/8_grpc_client.py (Python) and examples/grpc/ (Go/Rust/Node reference clients).

9. Web Dashboard

Monitor tasks, workers, metrics, and the DLQ through a built-in web UI — a FastAPI backend (conductor/api/) plus a React + Vite frontend (conductor/web/) with five screens:

1. Tasks — overview with status filter, free-text search, and pagination
2. Task details — payload, retry history, result, and error; cancel pending or retrying tasks
3. Workers — all registered workers with heartbeat, uptime, and counts
4. Metrics — Prometheus counters/gauges as hand-rolled SVG charts
5. Dead Letter — list DLQ tasks, retry or discard them

conductor api --host 0.0.0.0 --port 8080     # standalone dashboard server

Or embed it in a worker: Worker(api_enabled=True) / CONDUCTOR_API_ENABLED. Optional API-key auth (CONDUCTOR_API_KEY → X-API-Key header; unset = open). Tasks can be cancelled from the UI or the API (POST /api/tasks/{id}/cancel), which uses the new CANCELLED task status (schema v4). The built frontend is committed (conductor/web/dist) and ships in the wheel — no Node.js needed at install; rebuild with npm run build (see scripts/build_frontend.sh). See examples/9_web_dashboard.py.

10. Circuit Breaker

Protect your workers from repeatedly hammering a downstream service that is down. Enable a per-task-type circuit breaker on the worker:

from conductor import CircuitBreakerConfig, Worker

worker = Worker(
    database_url="...",
    circuit_breaker_enabled=True,
    circuit_breaker_config=CircuitBreakerConfig(
        threshold=5,          # open after 5 consecutive failures
        timeout=60.0,         # stay open 60s before probing
        half_open_attempts=2, # probe executions allowed while half-open
    ),
)

• After threshold consecutive failures of a task_type, the circuit trips OPEN — the worker skips that type (tasks stay pending, no false failures/retries/DLQ).
• After timeout seconds it becomes HALF_OPEN and lets a limited number of probe executions through; a successful probe closes the circuit, all probes failing re-opens it.
• Missing-handler errors never trip the breaker — only real handler exceptions.
• Configurable per worker via Worker(circuit_breaker_*) or the CONDUCTOR_CIRCUIT_BREAKER_* env vars. See examples/10_circuit_breaker.py.

11. Task Dependencies & Chaining

Chain tasks so they run in order automatically — a dependent task isn't picked up until its dependencies complete:

a = await queue.submit("download", {"url": "..."})
b = await queue.submit("process", {}, depends_on=[a])
c = await queue.submit("publish", {}, depends_on=[b])

• Gating — tasks with unmet dependencies stay pending (excluded from polling) until each dependency is completed (or cancelled).
• BLOCKED — when a dependency fails, its dependents are marked blocked (dependency '<id>' failed), propagating transitively (A→B→C).
• depends_on is preserved across DLQ retries (schema v5).
• Forward references are allowed; self-references are rejected. See examples/11_task_chaining.py.

12. Distributed Tracing (OpenTelemetry)

OpenTelemetry spans for the whole task lifecycle, shipped in an optional extra (pip install "conductor-task-queue[otel]"; without it every call is a no-op):

TRACING_ENABLED=true TRACING_EXPORTER=otlp \
  OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318/v1/traces \
  conductor worker --handlers myapp.handlers

• One trace per task — the submitter persists a W3C traceparent on the task row (schema v6) and the worker continues it, so a task executed by a different process still appears under the span that submitted it.
• Spans: conductor.task.submit / .submit_many / .execute / .cancel, conductor.dlq.retry / .discard, conductor.task.block_dependents, conductor.recurring.fire. Retry scheduling and DLQ moves show up as span events (retry.scheduled, task.dlq).
• Attribution without code changes: trace_id/span_id are added to every log record while a span is active.
• Worker.get_status() reports tracing_enabled / tracing_exporter.
• The gRPC API carries the context too (TaskRequest.traceparent).

See examples/12_distributed_tracing.py and docs/configuration.md.

---

Installation

Prerequisites

• Python 3.11+
• PostgreSQL 12+ — or MySQL 8.0+ / MariaDB 10.6+, or SQLite for embedded single-process deployments (pip install "conductor-task-queue[sqlite]")

Step 1: Install Conductor

pip install conductor-task-queue

# Embedded SQLite backend (no database server):
pip install "conductor-task-queue[sqlite]"

# MySQL/MariaDB backend (asyncmy driver):
pip install "conductor-task-queue[mysql]"

Step 2: Initialize Database

Conductor automatically creates tables on first connect. Just create the database and the schema is applied automatically:

createdb conductor

This creates:

• conductor_tasks – Task submissions and status
• conductor_workers – Worker heartbeats
• conductor_retries – Retry history
• conductor_dead_letter – Failed tasks

Note: A CLI migration tool is planned for a future release. For now, schema is auto-managed on first connect().

Step 3: Create a Worker

import asyncio
from conductor import Worker


async def main() -> None:
    async with Worker(
        database_url="postgresql://user:password@localhost/conductor"
    ) as worker:

        @worker.task("example_task")
        async def example_task(payload: dict) -> dict:
            print(f"Processing: {payload}")
            return {"status": "done"}

        await worker.run()


asyncio.run(main())

Run it:

python worker.py

Step 4: Submit Tasks

import asyncio
from conductor import TaskQueue


async def main() -> None:
    async with TaskQueue(
        database_url="postgresql://user:password@localhost/conductor"
    ) as queue:
        task_id = await queue.submit(
            task_type="example_task",
            payload={"message": "Hello, Conductor!"},
        )
        print(f"Submitted task: {task_id}")


asyncio.run(main())

Step 5: Run a Worker from the CLI (optional)

The package installs a conductor console script that runs a worker from environment variables (no application code needed):

# handlers.py exposes: def register(worker): @worker.task(...) ...
export DATABASE_URL=postgresql://user:password@localhost/conductor
conductor worker --handlers myapp.handlers

python -m conductor worker is equivalent. Run conductor --version to print the installed package version. See docs/installation.md for the handlers-module contract.

---

Usage Examples

Complete, runnable versions of these patterns live in examples/.

Example 1: Email Notifications

Send emails with automatic retry and error handling:

import asyncio
import logging

import aiohttp

from conductor import RetryPolicy, TaskQueue, Worker


logger = logging.getLogger(__name__)


async def main() -> None:
    async with Worker(
        database_url="postgresql://user:password@localhost/conductor"
    ) as worker:

        @worker.task("send_email")
        async def send_email(payload: dict) -> dict:
            """Send an email via SendGrid API."""
            to = payload["to"]
            subject = payload["subject"]
            body = payload["body"]

            logger.info("Sending email to %s", to)

            async with aiohttp.ClientSession() as session:
                async with session.post(
                    "https://api.sendgrid.com/v3/mail/send",
                    json={
                        "personalizations": [{"to": [{"email": to}]}],
                        "from": {"email": "noreply@example.com"},
                        "subject": subject,
                        "content": [{"type": "text/html", "value": body}],
                    },
                    headers={"Authorization": f"Bearer ..."},
                ) as resp:
                    resp.raise_for_status()

            logger.info("Email sent to %s", to)
            return {"status": "sent", "to": to}

        await worker.run()


# Submit an email task
async def submit_email() -> None:
    async with TaskQueue(
        database_url="postgresql://user:password@localhost/conductor"
    ) as queue:
        task_id = await queue.submit(
            task_type="send_email",
            payload={
                "to": "user@example.com",
                "subject": "Welcome!",
                "body": "<h1>Thanks for signing up</h1>",
            },
            retry_policy=RetryPolicy(
                max_retries=3,
                backoff_strategy="exponential",
                initial_delay=1.0,
            ),
        )
        print(f"Submitted task: {task_id}")


if __name__ == "__main__":
    asyncio.run(submit_email())

Example 2: Data Processing Pipeline

Multi-step data processing with task chaining:

import logging

from conductor import TaskQueue

logger = logging.getLogger(__name__)


@worker.task("process_upload")
async def process_upload(payload: dict) -> dict:
    """Process an uploaded file through a pipeline."""
    file_id = payload["file_id"]
    user_id = payload["user_id"]

    logger.info("Starting to process file %s", file_id)

    try:
        # Step 1: Download
        file_data = await download_file(file_id)

        # Step 2: Process
        result = await process_data(file_data)

        # Step 3: Store
        await store_result(file_id, result)

        # Step 4: Notify user (spawn another task)
        async with TaskQueue(
            database_url="postgresql://user:password@localhost/conductor"
        ) as queue:
            await queue.submit(
                task_type="send_notification",
                payload={
                    "user_id": user_id,
                    "message": "Your file is ready",
                    "result_id": result["id"],
                },
            )

        logger.info("Successfully processed file %s", file_id)
        return {"file_id": file_id, "status": "processed"}
    except Exception as e:
        logger.error("Failed to process file %s: %s", file_id, str(e))
        raise

Example 3: Scheduled Cleanup

Regular maintenance tasks:

import logging
from datetime import datetime, timezone

from conductor import TaskQueue

logger = logging.getLogger(__name__)


@worker.task("cleanup_expired_sessions")
async def cleanup_expired_sessions(payload: dict) -> dict:
    """Remove expired user sessions."""
    logger.info("Starting daily session cleanup")

    try:
        deleted_count = await db.execute(
            "DELETE FROM sessions WHERE expires_at < NOW()"
        )
        logger.info(
            "Deleted %d expired sessions", deleted_count
        )
        return {"deleted": deleted_count}
    except Exception as e:
        logger.error("Failed to cleanup sessions: %s", str(e))
        raise


# Submit via cron job: runs daily at 2 AM
# async with TaskQueue(database_url="...") as queue:
#     await queue.submit(
#         task_type="cleanup_expired_sessions",
#         payload={},
#     )

Tip: For one-off future runs use scheduled_for; for repeating (cron) runs use schedule_recurring() with the RecurringScheduler (or Worker(enable_scheduler=True)) instead of an external cron job.

Example 4: Error Handling & Idempotency

Patterns for robust task handlers:

@worker.task("process_order")
async def process_order(payload):
    """Process an order with idempotency"""
    order_id = payload["order_id"]
    
    # Check if already processed (idempotency)
    existing = await db.fetch_one(
        "SELECT id FROM processed_orders WHERE order_id = %s",
        order_id
    )
    
    if existing:
        logger.info(f"Order {order_id} already processed")
        return {"status": "already_processed", "order_id": order_id}
    
    try:
        # Process payment
        payment_result = await charge_payment(order_id)
        
        # Record as processed (atomic)
        await db.execute(
            "INSERT INTO processed_orders (order_id, payment_id) VALUES (%s, %s)",
            order_id,
            payment_result["id"]
        )
        
        logger.info(f"Order {order_id} processed successfully")
        return {"status": "success", "order_id": order_id}
    
    except PaymentError as e:
        # Specific error handling
        logger.error(f"Payment failed for order {order_id}: {e}")
        raise  # Retry
    
    except DatabaseError as e:
        # Database error - likely transient
        logger.error(f"Database error for order {order_id}: {e}")
        raise  # Retry
    
    except Exception as e:
        # Unexpected error
        logger.error(f"Unexpected error for order {order_id}: {e}", exc_info=True)
        raise  # Move to DLQ after retries

---

Configuration

Task Queue Options

from conductor import TaskQueue

queue = TaskQueue(
    database_url="postgresql://user:password@localhost/conductor",
    task_timeout=300.0,       # Task execution timeout (seconds)
    max_task_age=86400,       # Clean up tasks older than 1 day
    log_level="INFO",         # Logging level
    pool_min_size=2,          # Minimum DB pool connections
    pool_max_size=10,         # Maximum DB pool connections
    pool_timeout=30.0,        # DB pool acquire timeout
    command_timeout=60.0,     # SQL command timeout
)

Worker Options

from conductor import Worker

worker = Worker(
    database_url="postgresql://user:password@localhost/conductor",
    worker_id="worker-1",           # Unique worker identifier
    concurrency=10,                  # Max concurrent tasks
    poll_interval=0.5,               # Check for tasks every 500ms
    routes=None,                     # ``None`` polls all routes
    log_level="INFO",
    heartbeat_interval=10.0,         # DB heartbeat every 10s
    graceful_shutdown_timeout=30.0,  # Wait 30s for in-flight tasks
)

Retry Policies

from conductor import RetryPolicy

# Default retry policy
RetryPolicy(
    max_retries=3,
    backoff_strategy="exponential",
    initial_delay=1.0,  # seconds
    max_delay=3600.0,
)

# No retries
RetryPolicy(max_retries=0)

# Linear backoff (5s, 10s, 15s, 20s, 25s)
RetryPolicy(
    max_retries=5,
    backoff_strategy="linear",
    initial_delay=5.0,
    max_delay=3600.0,
)

# Fixed backoff (always 10s delay)
RetryPolicy(
    max_retries=10,
    backoff_strategy="fixed",
    initial_delay=10.0,
    max_delay=3600.0,
)

You can also pass a dict with matching keys (useful for JSON configs):

policy = RetryPolicy.from_dict({
    "max_retries": 3,
    "backoff_strategy": "exponential",
    "initial_delay": 1.0,
    "max_delay": 3600.0,
})

Environment Variables

# Database (required) — the scheme selects the backend:
#   postgresql://user:pass@host:5432/conductor   (default)
#   mysql://user:pass@host:3306/conductor        (MySQL 8.0+ / MariaDB 10.6+)
#   sqlite:///conductor.db                       (embedded, single process)
DATABASE_URL=postgresql://postgres:password@localhost:5432/conductor

# Worker
WORKER_ID=worker-1
CONCURRENCY=10
POLL_INTERVAL=0.5
ROUTES=default
HEARTBEAT_INTERVAL=10
GRACEFUL_SHUTDOWN_TIMEOUT=30
CONDUCTOR_HANDLERS_MODULE=

# Logging
LOG_LEVEL=INFO
LOG_FORMAT=json

# Metrics & Health (shared port)
METRICS_ENABLED=true
METRICS_PORT=8000
HEALTH_ENABLED=true

# Tasks
TASK_TIMEOUT=300
MAX_TASK_AGE=86400

# Connection pool
DB_MIN_SIZE=2
DB_MAX_SIZE=10
DB_TIMEOUT=30
DB_COMMAND_TIMEOUT=60
DB_BUSY_TIMEOUT=5

# Distributed tracing (requires the "otel" extra)
TRACING_ENABLED=false
TRACING_EXPORTER=otlp
OTEL_SERVICE_NAME=conductor
OTEL_EXPORTER_OTLP_ENDPOINT=
TRACING_SAMPLE_RATIO=1.0

Full reference: docs/configuration.md.

---

Deployment

Ready-to-use deployment artifacts are included in the repository and covered in detail in docs/deployment.md.

• Docker — a python:3.11-slim Dockerfile (non-root user, healthcheck on /health, ENTRYPOINT ["conductor"] CMD ["worker"]):

  docker build -t conductor:0.3.0 .
  docker run --rm -e DATABASE_URL=postgresql://... -p 8000:8000 conductor:0.3.0
  

• Docker Compose — docker-compose.yml (dev: PostgreSQL + worker) and docker-compose.prod.yml (replicas, resource limits, log rotation, nightly pg_dump backup):

  docker compose up -d --build
  docker compose -f docker-compose.prod.yml up -d --build --scale worker=3
  

• Kubernetes — examples/kubernetes.yaml (ConfigMap, Secret, Deployment with 3 replicas + liveness/readiness probes, Service):

  kubectl apply -f examples/kubernetes.yaml
  

• systemd — examples/conductor-worker.service (runs conductor worker):

  sudo cp examples/conductor-worker.service /etc/systemd/system/
  sudo systemctl daemon-reload && sudo systemctl enable --now conductor-worker
  

Validate deployment files locally without Docker:

python scripts/validate_deploy.py

The worker serves Prometheus metrics at /metrics and a JSON health check at /health on METRICS_PORT (default 8000). A Grafana dashboard is provided in docs/grafana/.

---

Architecture

How It Works

┌─────────────────────────────────────────────────────────┐
│                   Your Application                      │
│                  queue.submit(task)                     │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                  PostgreSQL Database                    │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │  Tasks   │  │ Workers  │  │ Retries  │  │  DLQ   │ │
│  └──────────┘  └──────────┘  └──────────┘  └────────┘ │
└──────────┬──────────────────────────────────┬───────────┘
           │                                  │
      Polls every │                          │
      500ms       ▼                          │
┌──────────────────────────────────────────────────────────┐
│                   Worker Processes                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ Worker-1 │  │ Worker-2 │  │ Worker-N │              │
│  │ async    │  │ async    │  │ async    │              │
│  └──────────┘  └──────────┘  └──────────┘              │
│  (Multiple machines or containers, all share DB)        │
└──────────────────────────────────────────────────────────┘

Key Design Decisions

1. The database is the source of truth – No separate message broker. All state lives in one place.
2. Polling-based task dispatch – Workers poll for new tasks every 500ms. No complex subscription logic.
3. Idempotent task processing – Workers record task IDs. Duplicate submissions are deduplicated automatically.
4. Async-first – Built on asyncio. No threads, no blocking I/O.
5. Observable – Every task transition logged and metered.
6. Portable across SQL backends – SQL is rendered through a per-backend dialect, so the only change between PostgreSQL, MySQL/MariaDB and SQLite is the DSN. PostgreSQL and MySQL/MariaDB claim rows with FOR UPDATE SKIP LOCKED and scale workers horizontally; SQLite is embedded and single-process (one worker per database file).

---

API Reference

TaskQueue

submit(task_type, payload, *, retry_policy=None, scheduled_for=None, route="default", priority=0, task_id=None)

Submit a new task (async).

Parameters:

• task_type (str): Logical type used to route the task to a handler
• payload (dict): Task data (JSON serializable)
• retry_policy (RetryPolicy, optional): Retry configuration
• scheduled_for (datetime, optional): Earliest pickup time
• route (str): Route for selective worker polling
• priority (int): Higher runs first
• task_id (str, optional): Explicit task ID

Returns: str – Task ID

Example:

task_id = await queue.submit(
    task_type="send_email",
    payload={"to": "user@example.com", "subject": "Hello"},
    retry_policy=RetryPolicy(max_retries=3, backoff_strategy="exponential"),
)

list_pending_tasks(limit=10, offset=0, route=None)

Get pending (not yet executed) tasks.

Parameters:

• limit (int): Max results
• offset (int): Pagination offset
• route (str | None): If set, only return tasks on this route

Returns: List[Task]

list_completed_tasks(limit=10, offset=0)

Get completed tasks.

list_failed_tasks(limit=10, offset=0)

Get tasks that failed (status='failed').

get_task(task_id)

Get a single task by ID.

Parameters:

• task_id (str): Task identifier

Returns: Task or None

Worker

@worker.task(task_type)

Decorator to register a task handler.

Parameters:

• task_type (str): Task type to handle

Example:

@worker.task("send_email")
async def handle_send_email(payload):
    # Process task
    return {"status": "sent"}

run()

Start the worker event loop (runs indefinitely).

Example:

asyncio.run(worker.run())

run_once()

Execute one polling cycle (useful for testing).

shutdown()

Gracefully shut down the worker.

DeadLetterQueue

list_tasks(limit=10, offset=0, include_discarded=False)

List tasks in the dead-letter queue.

Parameters:

• include_discarded (bool): Include soft-deleted tasks

Returns: List[DLQTask]

get_task(task_id)

Get a single DLQ task.

Returns: DLQTask or None

retry_task(task_id)

Retry a task from the DLQ (reset to pending, clears worker and error).

Returns: str — The task ID

discard_task(task_id, reason=None)

Permanently mark a DLQ task as discarded (soft-delete).

Parameters:

• reason (str, optional): Why the task is being discarded

count(include_discarded=False)

Count tasks in the dead-letter queue.

Returns: int

---

Best Practices

1. Design Idempotent Tasks

Always assume a task might run twice:

# Bad: Not idempotent
@worker.task("increment_counter")
async def increment_counter(payload):
    user_id = payload["user_id"]
    await db.execute("UPDATE users SET score = score + 1 WHERE id = ?", user_id)
    # If this runs twice, score increases by 2!

# Good: Idempotent
@worker.task("set_status")
async def set_status(payload):
    user_id = payload["user_id"]
    status = payload["status"]
    await db.execute("UPDATE users SET status = ? WHERE id = ?", status, user_id)
    # Safe to run multiple times (same result each time)

2. Use Structured Logging

Use lazy %-formatting for better performance:

import logging

logger = logging.getLogger(__name__)

@worker.task("process_data")
async def process_data(payload: dict) -> dict:
    task_id = payload.get("task_id", "unknown")
    logger.info("Task %s starting processing", task_id)
    # ... do work ...
    logger.info("Task %s completed", task_id)
    return {"status": "done"}

3. Handle Exceptions Gracefully

Always catch ConductorException for predictable error handling:

from conductor import TaskQueue
from conductor.exceptions import ConductorException


async with TaskQueue(database_url="...") as queue:
    try:
        task_id = await queue.submit(
            task_type="critical_job",
            payload={"data": "..."},
        )
        print(f"Submitted: {task_id}")
    except ConductorException as exc:
        # Database connection failure, invalid config, etc.
        print(f"Failed to submit: {exc}")

4. Monitor Dead Letter Queue

Don't ignore failed tasks:

import asyncio
from conductor import DeadLetterQueue


async def check_dlq() -> None:
    async with DeadLetterQueue(database_url="...") as dlq:
        count = await dlq.count()
        if count > 0:
            tasks = await dlq.list_tasks(limit=100)
            print(f"⚠️ {len(tasks)} tasks in DLQ")
            for task in tasks:
                print(f"  - {task.task_id}: {task.error_message}")


asyncio.run(check_dlq())

5. Scale Horizontally

Run multiple workers across machines:

# Machine 1
python worker.py --worker-id worker-1

# Machine 2
python worker.py --worker-id worker-2

# Machine 3
python worker.py --worker-id worker-3

All workers share the same PostgreSQL database. No coordination needed.

6. Use Environment Variables

Never hardcode credentials:

import os
from conductor import TaskQueue, Worker

database_url = os.getenv("DATABASE_URL")
queue = TaskQueue(database_url=database_url)
worker = Worker(database_url=database_url)

---

Troubleshooting

Issues with PostgreSQL Connection

Problem: ConnectionError: could not connect to server

Solutions:

1. Verify PostgreSQL is running: pg_isready -h localhost
2. Check connection string: DATABASE_URL=postgresql://user:password@host:port/database
3. Verify network permissions (firewall, security groups)
4. Check database credentials

Tasks not processing

Problem: Tasks submitted but not executed

Checklist:

1. Is the worker running?

   ps aux | grep worker.py
   

2. Are there tasks in the queue?

   from conductor import TaskQueue
   queue = TaskQueue(database_url="...")
   pending = queue.list_pending_tasks()
   print(f"Pending tasks: {len(pending)}")
   

3. Check worker logs:

   [2025-01-15 10:30:45] INFO: Worker started, listening for tasks
   [2025-01-15 10:30:46] DEBUG: Polled 0 tasks
   [2025-01-15 10:30:47] DEBUG: Polled 1 task (task_id: abc-123)
   

4. Verify task handler is registered:

   @worker.task("my_task")
   async def handler(payload):
       return {"status": "done"}
   

High latency

Problem: Tasks taking too long to execute after submission

Causes:

• Polling interval too long (default: 500ms)
• Worker overloaded (concurrency limit reached)
• Database queries slow

Solutions:

1. Increase polling frequency (trade-off: higher DB load):

   worker = Worker(poll_interval=0.1)  # Check every 100ms
   

2. Scale workers horizontally:

   python worker.py --worker-id worker-1 &
   python worker.py --worker-id worker-2 &
   python worker.py --worker-id worker-3 &
   

3. Increase worker concurrency:

   worker = Worker(concurrency=20)  # Handle 20 concurrent tasks
   

Database errors

Problem: DatabaseError: connection timeout or pool exhausted

Solutions:

1. Check PostgreSQL connection:

   psql postgresql://user:password@localhost/conductor -c "SELECT 1"
   

2. Schema is auto-managed — tables are created on first connect(). Verify with \dt in psql.

3. Check database load:

   SELECT count(*) FROM conductor_tasks WHERE status = 'pending';
   

4. Increase connection pool size:

   queue = TaskQueue(
       database_url="...",
       pool_min_size=5,
       pool_max_size=20,  # Increase from default 10
       pool_timeout=60.0,
   )
   

Worker crashes / signal handling

Problem: Worker doesn't shut down cleanly

Solutions:

1. Add signal handlers:

   import signal
   
   def handle_shutdown(sig, frame):
       print("Shutting down...")
       worker.shutdown()
   
   signal.signal(signal.SIGTERM, handle_shutdown)
   signal.signal(signal.SIGINT, handle_shutdown)
   

2. Increase graceful shutdown timeout:

   worker = Worker(
       database_url="...",
       graceful_shutdown_timeout=60  # Wait up to 60s for tasks
   )
   

---

Comparison to Alternatives

Feature	Conductor	Celery	RQ	Dramatiq
No external broker	✅	❌ (needs Redis/RabbitMQ)	❌ (needs Redis)	❌ (needs RabbitMQ)
Exactly-once semantics	✅	⚠️ (at-least-once)	⚠️ (at-least-once)	⚠️ (at-least-once)
Dead letter queue	✅	⚠️ (limited)	❌	✅
Built-in observability	✅	⚠️ (needs Flower)	⚠️ (needs tools)	⚠️ (needs tools)
Simple API	✅	❌ (steep learning curve)	✅	✅
Graceful shutdown	✅	✅	✅	✅
Scheduled tasks	⚠️ (v0.2)	✅	❌	✅
Async-native	✅	⚠️ (hybrid)	❌	✅
Runs on PostgreSQL, MySQL/MariaDB or SQLite	✅	⚠️ (PostgreSQL + Redis)	⚠️ (MySQL)	❌
Production-ready	✅	✅	⚠️	✅

When to use Conductor:

• You don't want Redis/RabbitMQ dependency
• You need reliability guarantees (exactly-once)
• You want simple deployment
• Observability is important

When to use alternatives:

• You already have Celery (switching is expensive)
• You need advanced routing (Celery)
• You have complex workflows (Airflow, Prefect)

---

Performance & Benchmarks

Benchmarks under typical conditions (PostgreSQL on localhost, 4 concurrent workers):

Metric	Target	Notes
Task submission	<2ms	Single-row insert
Task polling latency	~500ms	Dictated by poll_interval (configurable)
Task processing (empty handler)	<10ms	Status update + database write
Throughput (simple task)	400+ tasks/sec per worker	Scales linearly with worker count
Memory per worker	~50MB base	+ payload size
Database load (100 tasks/sec)	~15% CPU	PostgreSQL on localhost

Notes:

• Latency is dominated by polling interval (currently 500ms). Can be tuned for lower latency / higher DB load.
• Throughput scales linearly with worker count.
• Database is the bottleneck at very high scale (>10k tasks/sec). Consider sharding or dedicated task queue for extreme scale.

---

Roadmap

Phase 1 (v0.1 — Current)

✅ Completed:

• Basic task queue
• Retry logic with exponential backoff
• Dead letter queue with retry/discard API
• Worker pool with concurrency control
• Graceful shutdown with signal handling

🔄 In Progress:

• Structured logging (Sprint 5)
• Prometheus metrics exporter (Sprint 5)
• Health check endpoint (Sprint 5)
• Performance benchmarks (Sprint 6)
• Docker & deployment examples (Sprint 6)
• Comprehensive documentation (Sprint 6)

Phase 2 (v0.2.0 — released 2026-08-13)

✅ Task routing, priority queues, scheduled/recurring tasks (cron) ✅ Web dashboard, gRPC API ✅ Circuit breaker pattern, task dependencies/chaining

Phase 3 (v0.3.0 — released 2026-09-15; v0.4.0+ in progress)

✅ SQLite backend (v0.3.0) — embedded, single-process, parity-tested against PostgreSQL ✅ OpenTelemetry tracing (v0.3.0) — cross-process trace context (schema v6) ✅ MySQL/MariaDB backend (v0.3.0) — asyncmy driver, FOR UPDATE SKIP LOCKED row claiming 🔲 Advanced workflows, task versioning (v0.4.0)

---

Local Development

# 1. Start PostgreSQL
docker compose up -d

# 2. Copy and configure environment
cp .env.example .env

# 3. Create virtual environment & install
python3 -m venv .venv
source .venv/bin/activate
# Include the optional backends: the unit suite imports every backend module.
pip install -e ".[dev,sqlite,mysql,otel]"

# 4. Run tests (schema auto-creates on first connect)
pytest

# 5. Run linting and type checking
black .
mypy conductor/
flake8 conductor/ tests/

---

Contributing

We welcome contributions! See CONTRIBUTING.md for the full guide — development setup, how to run tests, code style, and the pull request process.

• 📄 CONTRIBUTING.md — how to contribute
• 🛡️ SECURITY.md — how to report a vulnerability
• 🤝 CODE_OF_CONDUCT.md — our community standards

Areas for Contribution

• Performance optimizations
• Additional retry strategies
• Observability enhancements
• Documentation improvements
• Example projects
• Database backend support (MariaDB tuning, additional backends)
• Polyglot worker implementations

---

Support & Community

• Documentation: docs.conductor.sh
• Issues & Bugs: GitHub Issues
• Discussions: GitHub Discussions

---

License

Conductor is licensed under the MIT License. See LICENSE file for details.

---

Authors

• Panagiotis Panageas (@Archangel-77) – Creator & Maintainer

---

Acknowledgments

Inspired by the need for a simpler task queue that doesn't require external infrastructure. Built for teams who value reliability, observability, and simplicity.

---

Start building reliable async systems with Conductor today.

pip install conductor-task-queue