Hooks and middleware for job events¶

Hooks run your code before a Soniq job, after it succeeds, and when it fails. Use them for logging, metrics, tracing and alerting without changing your job functions. Middleware wraps the whole job call when you need more control.

Registering hooks¶

Use the decorator methods on your Soniq instance:

soniq = Soniq(database_url="postgresql://localhost/myapp")

@soniq.before_job
async def log_start(job_name: str, job_id: str, attempt: int):
    print(f"Starting {job_name} ({job_id}), attempt {attempt}")

@soniq.after_job
async def log_success(job_name: str, job_id: str, duration_ms: float):
    print(f"Completed {job_name} ({job_id}) in {duration_ms}ms")

@soniq.on_error
async def log_failure(job_name: str, job_id: str, error: str, attempt: int):
    print(f"Failed {job_name} ({job_id}): {error} (attempt {attempt})")

You can register multiple hooks for the same event. They run in registration order. Soniq passes the arguments positionally, in the order shown below, so keep the argument order in your hook functions.

Hook signatures¶

@soniq.before_job¶

Called immediately before a job function executes.

async def before_job_hook(job_name: str, job_id: str, attempt: int) -> None

Argument	Type	Description
`job_name`	`str`	The registered job name (the `name=` you gave the decorator).
`job_id`	`str`	UUID of the job.
`attempt`	`int`	Current attempt number, starting at 1.

@soniq.after_job¶

Called after a job completes successfully.

async def after_job_hook(job_name: str, job_id: str, duration_ms: float) -> None

Argument	Type	Description
`job_name`	`str`	The registered job name.
`job_id`	`str`	UUID of the job.
`duration_ms`	`float`	Wall-clock execution time in milliseconds, rounded to 2 decimals.

@soniq.on_error¶

Called each time a job attempt fails, including the final attempt. Runs before the retry or dead-letter decision.

async def on_error_hook(job_name: str, job_id: str, error: str, attempt: int) -> None

Argument	Type	Description
`job_name`	`str`	The registered job name.
`job_id`	`str`	UUID of the job.
`error`	`str`	String representation of the exception.
`attempt`	`int`	The attempt number that failed.

Execution order¶

For a successful job:

before_job -> job function -> after_job

For a failed job:

before_job -> job function (raises) -> on_error

after_job and on_error are mutually exclusive for a given execution. before_job always runs, even if the job will ultimately fail.

A job that returns Snooze fires before_job but neither after_job nor on_error. A job whose stored arguments are corrupt goes to the dead-letter queue and fires no hooks. If you open a resource in before_job (such as a tracing span), do not rely on after_job or on_error alone to close it.

Sync and async hooks¶

Hooks can be either sync or async. Soniq inspects each hook at call time and awaits it if it is a coroutine function. A sync hook runs on the event loop, so keep it fast.

# Async hook
@soniq.before_job
async def async_hook(job_name, job_id, attempt):
    await some_async_operation()

# Sync hook works too
@soniq.before_job
def sync_hook(job_name, job_id, attempt):
    some_sync_operation()

Error handling in hooks¶

If a hook raises an exception, Soniq logs a warning and continues. A broken hook never prevents a job from running or its result from being recorded.

WARNING - Hook before_job failed: ConnectionError(...)

The log message has the form Hook <event> failed: <exception>.

This means hooks are safe for non-critical operations like metrics and logging. If you need a hook failure to stop job execution, handle that logic inside the job function itself.

Practical examples¶

Structured logging¶

import logging
import json

logger = logging.getLogger("soniq.hooks")

@soniq.before_job
async def structured_log_start(job_name, job_id, attempt):
    logger.info(json.dumps({
        "event": "job_started",
        "job_name": job_name,
        "job_id": job_id,
        "attempt": attempt,
    }))

@soniq.after_job
async def structured_log_done(job_name, job_id, duration_ms):
    logger.info(json.dumps({
        "event": "job_completed",
        "job_name": job_name,
        "job_id": job_id,
        "duration_ms": duration_ms,
    }))

Prometheus metrics¶

from prometheus_client import Counter, Histogram

jobs_started = Counter("soniq_jobs_started", "Jobs started", ["job_name"])
jobs_completed = Histogram("soniq_jobs_duration_ms", "Job duration", ["job_name"])
jobs_failed = Counter("soniq_jobs_failed", "Jobs failed", ["job_name"])

@soniq.before_job
async def track_start(job_name, job_id, attempt):
    jobs_started.labels(job_name=job_name).inc()

@soniq.after_job
async def track_duration(job_name, job_id, duration_ms):
    jobs_completed.labels(job_name=job_name).observe(duration_ms)

@soniq.on_error
async def track_failure(job_name, job_id, error, attempt):
    jobs_failed.labels(job_name=job_name).inc()

Slack alerting on dead-letter¶

@soniq.on_error
async def alert_on_final_failure(job_name, job_id, error, attempt):
    # Look up the job's max_attempts from the registry
    job_meta = soniq.registry.get_job(job_name)
    if job_meta and attempt >= job_meta["max_retries"] + 1:
        await send_slack_message(
            f"Job {job_name} ({job_id}) moved to dead-letter queue: {error}"
        )

OpenTelemetry tracing¶

from opentelemetry import trace

tracer = trace.get_tracer("soniq")
_spans = {}

@soniq.before_job
async def start_span(job_name, job_id, attempt):
    span = tracer.start_span(f"job:{job_name}", attributes={
        "job.id": job_id,
        "job.attempt": attempt,
    })
    _spans[job_id] = span

@soniq.after_job
async def end_span_success(job_name, job_id, duration_ms):
    span = _spans.pop(job_id, None)
    if span:
        span.set_attribute("job.duration_ms", duration_ms)
        span.end()

@soniq.on_error
async def end_span_error(job_name, job_id, error, attempt):
    span = _spans.pop(job_id, None)
    if span:
        span.set_status(trace.StatusCode.ERROR, error)
        span.end()

Middleware¶

Middleware wraps every job call. Register it with @soniq.middleware. The first middleware you register is the outermost.

import time
from soniq import JobContext

@soniq.middleware
async def timing(ctx: JobContext, call_next):
    started = time.monotonic()
    try:
        return await call_next(ctx)
    finally:
        print(f"{ctx.job_name} took {time.monotonic() - started:.2f}s")

Call await call_next(ctx) to continue the chain, and return its result.

Next steps¶

Jobs API - JobContext, Snooze and retry options.
Custom metrics sink - send job metrics without hooks.