Events

flexiq emits a lifecycle event for everything that happens to a job, a worker, a queue, or a workflow run. The in-process event bus dispatches those to Python callbacks in the same process: lowest latency, no serialization, no HTTP.

For delivery to an external service — with persistence, signing, and a replay log — use webhooks, which are driven by the same event set.

Event types

The EventType enum defines all available lifecycle events. Each member's .value is the dotted, lowercase wire string used in webhook payloads, delivery logs, and the REST API — e.g. EventType.JOB_FAILED.value == "job.failed":

duration_ms is how long the job ran, so a listener needn't time it. On the outcome events (JOB_RETRYING, JOB_DEAD, JOB_CANCELLED) it is None when nothing measured the run — a job that failed before it ever executed.

In-process listeners

Use queue.on_event() to subscribe a callback. Callbacks run in a ThreadPoolExecutor so they never block the worker, and exceptions are logged but don't affect job processing.

from flexiq import Queue
from flexiq.events import EventType

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

def on_failure(event_type: EventType, payload: dict) -> None:
    print(f"Job {payload['job_id']} failed: {payload.get('error')}")

queue.on_event(EventType.JOB_FAILED, on_failure)

Configure the pool size via Queue(event_workers=N) (default 4) if your callbacks are slow.

Event ordering

Events fire in the order the scheduler processes results — typically the order jobs complete. For jobs that complete nearly simultaneously, ordering is not guaranteed across different workers or threads.

Within a single job's lifecycle, events always fire in this order:

1. JOB_ENQUEUED (at enqueue time)
2. JOB_COMPLETED / JOB_FAILED / JOB_CANCELLED (at completion)
3. JOB_RETRYING (if retried, before the next attempt)
4. JOB_DEAD (if all retries exhausted)

Examples

Slack on failure

When the receiver is the same Python process and you don't need persistence, the event bus is the cheapest path:

import requests
from flexiq.events import EventType

def notify_slack(event_type: EventType, payload: dict) -> None:
    requests.post(
        "https://hooks.slack.com/services/T.../B.../xxx",
        json={
            "text": f":x: Task `{payload['task_name']}` failed\n"
                    f"Job ID: `{payload['job_id']}`\n"
                    f"Error: {payload.get('error', 'unknown')}"
        },
    )

queue.on_event(EventType.JOB_FAILED, notify_slack)
queue.on_event(EventType.JOB_DEAD, notify_slack)

Database audit trail

from flexiq.events import EventType

def audit_log(event_type: EventType, payload: dict) -> None:
    db.execute(
        "INSERT INTO audit_log (event, job_id, task_name, timestamp) VALUES (?, ?, ?, ?)",
        (event_type.value, payload["job_id"], payload["task_name"], time.time()),
    )

for event in [
    EventType.JOB_ENQUEUED,
    EventType.JOB_COMPLETED,
    EventType.JOB_FAILED,
    EventType.JOB_DEAD,
]:
    queue.on_event(event, audit_log)