Callback Patterns

A callback is a function passed as an argument to another function, to be called later. Callbacks are fundamental to event-driven programming, asynchronous operations, and flexible APIs.

---

What is a Callback?

def process_data(data, on_complete):
    """Process data and call on_complete when done."""
    result = [x * 2 for x in data]
    on_complete(result)  # Callback invocation

def print_result(result):
    print(f"Processing complete: {result}")

# Pass function as callback
process_data([1, 2, 3], print_result)
# Output: Processing complete: [2, 4, 6]

The key insight: functions are first-class objects in Python, so they can be passed around like any other value.

---

Basic Callback Patterns

Simple Callback

def fetch_data(url, callback):
    """Simulate fetching data and calling back with result."""
    # Simulate network request
    data = {"status": "ok", "url": url}
    callback(data)

def handle_response(data):
    print(f"Received: {data}")

fetch_data("https://api.example.com", handle_response)

Callback with Lambda

def calculate(x, y, operation):
    """Apply operation callback to x and y."""
    return operation(x, y)

# Using lambdas as callbacks
calculate(10, 5, lambda a, b: a + b)  # 15
calculate(10, 5, lambda a, b: a * b)  # 50
calculate(10, 5, lambda a, b: a ** b) # 100000

Multiple Callbacks

def process(data, on_success, on_error):
    """Process with separate success and error callbacks."""
    try:
        result = [x * 2 for x in data]
        on_success(result)
    except Exception as e:
        on_error(e)

def success_handler(result):
    print(f"Success: {result}")

def error_handler(error):
    print(f"Error: {error}")

process([1, 2, 3], success_handler, error_handler)
# Success: [2, 4, 6]

process("not a list", success_handler, error_handler)
# Error: can't multiply sequence by non-int

---

Callback with Context

Passing Extra Arguments

from functools import partial

def notify(message, callback, **context):
    """Notify with additional context."""
    callback(message, **context)

def log_message(message, level="INFO", timestamp=None):
    print(f"[{level}] {timestamp}: {message}")

notify("Server started", log_message, level="INFO", timestamp="10:30")
# [INFO] 10:30: Server started

Using partial for Context

from functools import partial

def send_email(to, subject, body):
    print(f"To: {to}\nSubject: {subject}\n{body}")

def process_order(order_id, on_complete):
    # Process order...
    on_complete(order_id)

# Create callback with pre-filled arguments
email_callback = partial(
    send_email,
    to="customer@example.com",
    subject="Order Confirmation"
)

def notify_order(order_id):
    email_callback(body=f"Your order #{order_id} is confirmed!")

process_order(12345, notify_order)

Closure for Context

def create_logger(prefix):
    """Create a callback with embedded context."""
    def log(message):
        print(f"[{prefix}] {message}")
    return log

debug_log = create_logger("DEBUG")
error_log = create_logger("ERROR")

def process(data, logger):
    logger(f"Processing {len(data)} items")
    # ... process data
    logger("Complete")

process([1, 2, 3], debug_log)
# [DEBUG] Processing 3 items
# [DEBUG] Complete

---

Event Handler Pattern

Simple Event System

class EventEmitter:
    def __init__(self):
        self._callbacks = {}

    def on(self, event, callback):
        """Register a callback for an event."""
        if event not in self._callbacks:
            self._callbacks[event] = []
        self._callbacks[event].append(callback)

    def emit(self, event, *args, **kwargs):
        """Trigger all callbacks for an event."""
        for callback in self._callbacks.get(event, []):
            callback(*args, **kwargs)

# Usage
emitter = EventEmitter()

def on_user_login(user):
    print(f"User logged in: {user}")

def on_user_login_log(user):
    print(f"Logging: {user} login at {datetime.now()}")

emitter.on("login", on_user_login)
emitter.on("login", on_user_login_log)

emitter.emit("login", "alice")
# User logged in: alice
# Logging: alice login at 2024-01-15 10:30:00

With Decorator Registration

class EventSystem:
    _handlers = {}

    @classmethod
    def on(cls, event):
        """Decorator to register event handler."""
        def decorator(func):
            if event not in cls._handlers:
                cls._handlers[event] = []
            cls._handlers[event].append(func)
            return func
        return decorator

    @classmethod
    def emit(cls, event, *args, **kwargs):
        for handler in cls._handlers.get(event, []):
            handler(*args, **kwargs)

# Register handlers with decorator
@EventSystem.on("order_placed")
def send_confirmation(order):
    print(f"Sending confirmation for order {order['id']}")

@EventSystem.on("order_placed")
def update_inventory(order):
    print(f"Updating inventory for {order['items']}")

# Trigger event
EventSystem.emit("order_placed", {"id": 123, "items": ["book", "pen"]})

---

Observer Pattern

class Subject:
    """Observable that notifies observers of state changes."""

    def __init__(self):
        self._observers = []
        self._state = None

    def attach(self, observer):
        """Add an observer callback."""
        self._observers.append(observer)

    def detach(self, observer):
        """Remove an observer callback."""
        self._observers.remove(observer)

    def notify(self):
        """Notify all observers of state change."""
        for observer in self._observers:
            observer(self._state)

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, value):
        self._state = value
        self.notify()

# Usage
def observer_a(state):
    print(f"Observer A: state changed to {state}")

def observer_b(state):
    print(f"Observer B: state changed to {state}")

subject = Subject()
subject.attach(observer_a)
subject.attach(observer_b)

subject.state = "active"
# Observer A: state changed to active
# Observer B: state changed to active

subject.detach(observer_a)
subject.state = "inactive"
# Observer B: state changed to inactive

---

Progress Callback Pattern

def process_items(items, on_progress=None):
    """Process items with optional progress callback."""
    results = []
    total = len(items)

    for i, item in enumerate(items):
        # Process item
        result = item * 2
        results.append(result)

        # Report progress
        if on_progress:
            progress = (i + 1) / total * 100
            on_progress(progress, item, result)

    return results

def print_progress(percent, item, result):
    print(f"{percent:.0f}% - Processed {item} -> {result}")

process_items([1, 2, 3, 4, 5], on_progress=print_progress)
# 20% - Processed 1 -> 2
# 40% - Processed 2 -> 4
# 60% - Processed 3 -> 6
# 80% - Processed 4 -> 8
# 100% - Processed 5 -> 10

Progress with Cancellation

def process_items(items, on_progress=None):
    """Process items; callback returns False to cancel."""
    results = []
    total = len(items)

    for i, item in enumerate(items):
        result = item * 2
        results.append(result)

        if on_progress:
            progress = (i + 1) / total * 100
            should_continue = on_progress(progress, item, result)
            if should_continue is False:
                print("Processing cancelled")
                break

    return results

def limited_progress(percent, item, result):
    print(f"{percent:.0f}%")
    return percent < 50  # Cancel after 50%

process_items([1, 2, 3, 4, 5], on_progress=limited_progress)
# 20%
# 40%
# 60%
# Processing cancelled

---

Retry with Callback

import time
import random

def retry_operation(operation, max_attempts=3, on_retry=None):
    """Retry operation with callback on each retry."""
    last_error = None

    for attempt in range(1, max_attempts + 1):
        try:
            return operation()
        except Exception as e:
            last_error = e
            if on_retry and attempt < max_attempts:
                on_retry(attempt, e)
            time.sleep(0.1 * attempt)  # Exponential backoff

    raise last_error

def unreliable_operation():
    if random.random() < 0.7:
        raise ConnectionError("Network timeout")
    return "Success!"

def on_retry(attempt, error):
    print(f"Attempt {attempt} failed: {error}. Retrying...")

result = retry_operation(unreliable_operation, max_attempts=5, on_retry=on_retry)

---

Validation Callback Pattern

def create_user(username, email, validators=None):
    """Create user with validation callbacks."""
    validators = validators or []

    data = {"username": username, "email": email}

    # Run all validators
    errors = []
    for validator in validators:
        error = validator(data)
        if error:
            errors.append(error)

    if errors:
        raise ValueError(f"Validation failed: {errors}")

    return data

# Validator callbacks
def validate_username(data):
    if len(data["username"]) < 3:
        return "Username must be at least 3 characters"

def validate_email(data):
    if "@" not in data["email"]:
        return "Invalid email format"

def validate_no_spaces(data):
    if " " in data["username"]:
        return "Username cannot contain spaces"

# Usage
validators = [validate_username, validate_email, validate_no_spaces]

try:
    user = create_user("ab", "invalid", validators)
except ValueError as e:
    print(e)
# Validation failed: ['Username must be at least 3 characters', 'Invalid email format']

---

Middleware Pattern

def apply_middleware(data, middlewares):
    """Apply chain of middleware callbacks."""
    result = data
    for middleware in middlewares:
        result = middleware(result)
        if result is None:
            break  # Middleware can halt chain
    return result

# Middleware callbacks
def add_timestamp(data):
    data["timestamp"] = "2024-01-15"
    return data

def add_user_id(data):
    data["user_id"] = 123
    return data

def validate_data(data):
    if "name" not in data:
        print("Validation failed: missing name")
        return None  # Halt chain
    return data

# Apply middleware chain
middlewares = [add_timestamp, add_user_id, validate_data]

result = apply_middleware({"name": "test"}, middlewares)
print(result)
# {'name': 'test', 'timestamp': '2024-01-15', 'user_id': 123}

result = apply_middleware({}, middlewares)
# Validation failed: missing name
# None

---

Async Callback Simulation

import threading
import time

def async_fetch(url, on_complete, on_error=None):
    """Simulate async operation with callbacks."""
    def do_fetch():
        try:
            time.sleep(1)  # Simulate network delay
            data = {"url": url, "status": "ok"}
            on_complete(data)
        except Exception as e:
            if on_error:
                on_error(e)

    thread = threading.Thread(target=do_fetch)
    thread.start()
    return thread

def handle_result(data):
    print(f"Received: {data}")

def handle_error(error):
    print(f"Error: {error}")

print("Starting async fetch...")
thread = async_fetch("https://api.example.com", handle_result, handle_error)
print("Continuing with other work...")
thread.join()  # Wait for completion
# Starting async fetch...
# Continuing with other work...
# (1 second later)
# Received: {'url': 'https://api.example.com', 'status': 'ok'}

---

Best Practices

Use Type Hints

from typing import Callable, Optional

def process(
    data: list,
    on_complete: Callable[[list], None],
    on_error: Optional[Callable[[Exception], None]] = None
) -> None:
    """Process with typed callbacks."""
    try:
        result = [x * 2 for x in data]
        on_complete(result)
    except Exception as e:
        if on_error:
            on_error(e)

Default No-Op Callback

def process(data, callback=None):
    """Use no-op default instead of None check."""
    callback = callback or (lambda x: None)
    result = [x * 2 for x in data]
    callback(result)
    return result

Document Callback Signatures

def fetch_data(url, callback):
    """
    Fetch data from URL.

    Args:
        url: The URL to fetch
        callback: Function called with (data, error) where:
            - data: The response dict if successful, None on error
            - error: Exception if failed, None on success
    """
    pass

---

Summary

Pattern	Use Case
Simple callback	Basic async completion
Success/Error callbacks	Error handling
Event emitter	Multiple listeners
Observer	State change notifications
Progress callback	Long-running operations
Validation callbacks	Pluggable validation
Middleware	Request/response processing

Key Takeaways:

• Callbacks enable flexible, extensible APIs
• Use multiple callbacks for success/error handling
• functools.partial and closures provide context
• Event systems allow multiple subscribers
• Progress callbacks should support cancellation
• Type hints document expected callback signatures
• Consider async/await for complex async patterns