Skip to content

Job Middleware and Hook Patterns

DodaTech Updated 2026-06-28 7 min read

In this tutorial, you will learn about Job Middleware and Hook Patterns. We cover key concepts, practical examples, and best practices to help you master this topic.

Implement middleware hooks for background job processing including before/after hooks, around filters, error hooks, and middleware pipelines for cross-cutting concerns.

What You Learn

You will learn how to build a middleware pipeline for job processing, implement before/after hooks for logging and metrics, create error hooks for failure handling, and compose middleware modules.

Why It Matters

Cross-cutting concerns like logging, metrics, authentication, and Rate Limiting should not be duplicated in every job. Middleware hooks apply them consistently across all jobs without code duplication.

Real-World Use

DodaTech's job system uses middleware for: logging every job execution, collecting duration metrics, tracking rate limits per user, and automatically retrying transient failures. Each middleware is a reusable module.

Middleware Pipeline Architecture

flowchart TD
    J[Job Arrives] --> M1[Metrics Middleware]
    M1 --> M2[Logging Middleware]
    M2 --> M3[Auth Middleware]
    M3 --> M4[Rate Limit Middleware]
    M4 --> H[Job Handler]
    H --> M5[Cleanup Middleware]
    M5 --> E[Complete]
    H -.->|Error| EH[Error Middleware]
    EH --> E

Middleware Pipeline

import time
import functools

class MiddlewarePipeline:
    def __init__(self):
        self._middlewares = []
        self._handlers = {}

    def use(self, middleware):
        self._middlewares.append(middleware)
        return self

    def handler(self, name):
        def decorator(func):
            self._handlers[name] = func
            return func
        return decorator

    def execute(self, name, **context):
        handler = self._handlers.get(name)
        if not handler:
            raise ValueError(f"Unknown handler: {name}")

        def create_chain(middlewares, final_handler):
            if not middlewares:
                return final_handler
            mw = middlewares[0]
            rest = middlewares[1:]
            return lambda ctx: mw(ctx, create_chain(rest, final_handler))

        chain = create_chain(self._middlewares, handler)
        return chain(context)

class MetricsMiddleware:
    def __call__(self, context, next_middleware):
        start = time.time()
        context['metrics'] = {'started_at': start}
        try:
            result = next_middleware(context)
            duration = time.time() - start
            context['metrics']['duration'] = duration
            print(f"  Metrics: {context.get('job_type', 'job')} took {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            context['metrics']['duration'] = duration
            context['metrics']['error'] = str(e)
            print(f"  Metrics: {context.get('job_type', 'job')} FAILED after {duration:.2f}s")
            raise

class LoggingMiddleware:
    def __call__(self, context, next_middleware):
        print(f"  Log: Starting job {context.get('job_id', 'unknown')}")
        try:
            result = next_middleware(context)
            print(f"  Log: Completed job {context.get('job_id', 'unknown')}")
            return result
        except Exception as e:
            print(f"  Log: Failed job {context.get('job_id', 'unknown')}: {e}")
            raise

pipeline = MiddlewarePipeline()

@pipeline.handler('scan')
def scan_job(context):
    print(f"  Scanning: {context['file']}")
    time.sleep(0.2)
    return {'threats': 0}

pipeline.use(MetricsMiddleware())
pipeline.use(LoggingMiddleware())

result = pipeline.execute('scan', job_id='scan-001', job_type='file_scan',
                          file='document.pdf')
print(f"Result: {result}")

Expected output:

  Log: Starting job scan-001
  Scanning: document.pdf
  Log: Completed job scan-001
  Metrics: file_scan took 0.20s
Result: {'threats': 0}

Before/After Hooks

import time

class HookedJob:
    def __init__(self):
        self.before_hooks = []
        self.after_hooks = []
        self.error_hooks = []

    def before(self, func):
        self.before_hooks.append(func)
        return func

    def after(self, func):
        self.after_hooks.append(func)
        return func

    def on_error(self, func):
        self.error_hooks.append(func)
        return func

    def execute(self, job_name, job_func, **kwargs):
        context = {'job_name': job_name, 'started_at': time.time(), **kwargs}

        for hook in self.before_hooks:
            hook(context)

        try:
            result = job_func(context)
            context['result'] = result
            for hook in self.after_hooks:
                hook(context)
            return result
        except Exception as e:
            context['error'] = str(e)
            for hook in self.error_hooks:
                hook(context)
            raise

hooked = HookedJob()

@hooked.before
def log_start(ctx):
    print(f"[{ctx['job_name']}] Starting at {time.strftime('%H:%M:%S')}")

@hooked.before
def increment_counter(ctx):
    print(f"[{ctx['job_name']}] Counter incremented")

@hooked.after
def log_duration(ctx):
    duration = time.time() - ctx['started_at']
    print(f"[{ctx['job_name']}] Completed in {duration:.2f}s")

@hooked.on_error
def log_error(ctx):
    print(f"[{ctx['job_name']}] FAILED: {ctx['error']}")

def process_job(ctx):
    print(f"  Processing {ctx.get('file', 'unknown')}")
    time.sleep(0.1)
    return {'status': 'ok'}

hooked.execute('file_scan', process_job, file='report.pdf')

Expected output:

[file_scan] Starting at 10:00:00
[file_scan] Counter incremented
  Processing report.pdf
[file_scan] Completed in 0.10s

Around Filters

import time

class AroundFilter:
    def __init__(self):
        self.filters = []

    def add(self, name, filter_func):
        self.filters.append((name, filter_func))
        return self

    def execute(self, job_func, **kwargs):
        chain = job_func
        for name, filter_func in reversed(self.filters):
            def make_filter(fn, fname, ff):
                def filtered(**kw):
                    print(f"  [{fname}] before")
                    start = time.time()
                    try:
                        result = fn(**kw)
                        duration = time.time() - start
                        print(f"  [{fname}] after ({duration:.2f}s)")
                        return result
                    except Exception as e:
                        print(f"  [{fname}] error: {e}")
                        raise
                return filtered
            chain = make_filter(chain, name, filter_func)

        return chain(**kwargs)

around = AroundFilter()

def db_transaction(next_func, **kwargs):
    print("  [DB] Transaction started")
    try:
        result = next_func(**kwargs)
        print("  [DB] Transaction committed")
        return result
    except Exception as e:
        print("  [DB] Transaction rolled back")
        raise

def timing(next_func, **kwargs):
    start = time.time()
    result = next_func(**kwargs)
    print(f"  [Timing] Duration: {time.time() - start:.2f}s")
    return result

around.add('db', db_transaction).add('timing', timing)

def process_payment(amount, account):
    print(f"  Processing payment ${amount} to {account}")
    return {'txn_id': 'txn_001'}

result = around.execute(process_payment, amount=100, account='acc-123')
print(f"Result: {result}")

Expected output:

  [db] before
  [timing] before
  Processing payment $100 to acc-123
  [timing] after (0.00s)
  [db] after (0.00s)
Result: {'txn_id': 'txn_001'}

Error Handling Middleware

import time

class ErrorHandlingMiddleware:
    def __init__(self):
        self.error_handlers = {}

    def catch(self, exception_type):
        def decorator(func):
            self.error_handlers[exception_type] = func
            return func
        return decorator

    def execute(self, job_func, **kwargs):
        try:
            return job_func(**kwargs)
        except tuple(self.error_handlers.keys()) as e:
            handler = self.error_handlers[type(e)]
            result = handler(e, kwargs)
            print(f"  Handled {type(e).__name__}: {e}")
            return result

class NetworkError(Exception):
    pass

class TimeoutError(Exception):
    pass

def process_data(data):
    if data.get('type') == 'timeout':
        raise TimeoutError("Connection timed out")
    if data.get('type') == 'network':
        raise NetworkError("Network unreachable")
    return {'processed': data}

ehm = ErrorHandlingMiddleware()

@ehm.catch(NetworkError)
def handle_network(err, context):
    print(f"  Retrying network operation...")
    return {'status': 'retried'}

@ehm.catch(TimeoutError)
def handle_timeout(err, context):
    print(f"  Scheduling retry for later...")
    return {'status': 'scheduled_retry'}

result1 = ehm.execute(process_data, data={'type': 'network'})
result2 = ehm.execute(process_data, data={'type': 'timeout'})
print(f"Results: {result1}, {result2}")

Expected output:

  Retrying network operation...
  Handled NetworkError: Network unreachable
  Scheduling retry for later...
  Handled TimeoutError: Connection timed out
Results: {'status': 'retried'}, {'status': 'scheduled_retry'}

Common Mistakes

1. Middleware State Leakage

Middleware that modifies shared state across jobs causes race conditions. Each middleware invocation should use clean context.

2. Order-Dependent Middleware

Relying on middleware execution order creates hidden coupling. Document the expected order and validate it at startup.

3. Error Swallowing

Middleware that catches all exceptions without re-raising can hide critical failures. Re-raise after handling or logging.

4. Heavy Middleware

Slow middleware (external API calls, heavy computations) blocks the entire pipeline. Keep middleware fast or offload heavy work.

5. No Middleware Testing

Middleware affects all jobs. A bug in middleware breaks every job. Test middleware in isolation and integration.

Practice Questions

1. What is the middleware pattern in job processing?

A pipeline of functions that wrap job execution. Each middleware handles a cross-cutting concern and calls the next middleware or final handler.

2. How do before/after hooks differ from around filters?

Before/after hooks run at specific points. Around filters wrap the entire execution and can modify input and output.

3. Why is middleware order important?

Some middleware depends on context set by earlier middleware. Metrics middleware must wrap everything to measure total duration.

4. How do error hooks work in middleware?

When the handler raises an exception, error hooks run instead of after hooks. They can log, cleanup, or provide fallback results.

Challenge

Build a middleware system for a payment processing queue with: logging middleware (log every job), metrics middleware (duration, success/failure), retry middleware (auto-retry transient failures), rate limit middleware (per-user limits), and Transaction middleware (DB commit/rollback).

FAQ

Can middleware modify job data?

Yes. Middleware can add context, modify inputs, or transform outputs. This is useful for enrichment or sanitization.

How do I skip middleware for specific jobs?

Add a 'skip_middleware' flag in job context. Each middleware checks the flag and passes through if set.

Is middleware re-entrant?

Yes. Each job execution creates a fresh middleware chain. No state is shared between different job executions.

Can middleware be async?

Yes, for async job processors. Use async middleware functions that await the next middleware in the chain.

How do I debug middleware issues?

Add debug middleware that logs every call and context change. Place it first in the chain to capture all activity.

Mini Project: Middleware Pipeline

import time

class MiddlewarePipeline:
    def __init__(self):
        self.layers = []

    def add(self, name, before=None, after=None, error=None):
        self.layers.append({
            'name': name, 'before': before, 'after': after, 'error': error
        })

    def run(self, handler, context):
        context['_start'] = time.time()

        for layer in self.layers:
            if layer['before']:
                layer['before'](context)

        try:
            result = handler(context)
            context['result'] = result
        except Exception as e:
            context['error'] = str(e)
            for layer in reversed(self.layers):
                if layer['error']:
                    layer['error'](context)
            raise

        for layer in reversed(self.layers):
            if layer['after']:
                layer['after'](context)

        return context.get('result')

pipe = MiddlewarePipeline()
pipe.add('logging',
         before=lambda c: print(f"Start {c['name']}"),
         after=lambda c: print(f"End {c['name']} ({time.time()-c['_start']:.2f}s)"),
         error=lambda c: print(f"Error {c['name']}: {c['error']}"))
pipe.add('metrics',
         after=lambda c: print(f"  Duration: {time.time()-c['_start']:.2f}s"))

pipe.run(lambda c: time.sleep(0.1) or 'done', {'name': 'test_job'})

Expected output:

Start test_job
End test_job (0.10s)
  Duration: 0.10s

What's Next

Now that you understand middleware, explore job lifecycle events for event-driven processing, then learn about job metrics with Prometheus for monitoring.

← Previous Job Timeout Configuration and Handling Next → Job Lifecycle Events and Hooks

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Background Jobs