Callbacks and Higher-Order Functions

Introduction

Python supports higher-order functions — functions that can receive other functions as arguments or return functions. This is similar to JavaScript's callback pattern but Python offers additional approaches for asynchronous programming.

Callbacks in Python are commonly used for:

• Event handling
• Functional programming patterns
• Asynchronous programming (with asyncio)
• Threading operations
• Processing pipelines

Unlike JavaScript's event loop, Python has explicit approaches to asynchronous programming including threading, asyncio, and concurrent.futures.

---

Concepts

Higher-Order Functions

• Functions that take other functions as parameters
• Functions that return other functions
• Built-in examples: map(), filter(), sorted()

Callback Patterns

• Pass functions to be called later
• Common in event-driven programming
• Used with threading and async operations

Functional Programming

• Functions as first-class objects
• Lambda expressions for simple callbacks
• Decorators as callback-like patterns

---

Examples

Basic Callback Pattern

import time
import threading

def process_data(data, callback):
    """Process data and call callback with result"""
    print(f"Processing {data}...")
  
    # Simulate processing time
    def delayed_processing():
        time.sleep(2)
        result = data.upper()
        callback(result)
  
    # Start processing in background thread
    thread = threading.Thread(target=delayed_processing)
    thread.start()

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

# Usage
process_data("hello world", handle_result)
print("Program continues...")
# Output order:
# Processing hello world...
# Program continues...
# Processing complete: HELLO WORLD

Using Lambda Functions

def calculate_async(x, y, operation, callback):
    """Perform calculation and call callback"""
    def worker():
        time.sleep(1)  # Simulate delay
        if operation == "add":
            result = x + y
        elif operation == "multiply":
            result = x * y
        elif operation == "power":
            result = x ** y
        else:
            result = None
        callback(result)
  
    threading.Thread(target=worker).start()

# Usage with lambda
calculate_async(5, 3, "multiply", lambda result: print(f"Result: {result}"))
calculate_async(2, 8, "power", lambda result: print(f"Power: {result}"))

Built-in Higher-Order Functions

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

# map() applies function to each element
squares = list(map(lambda x: x**2, numbers))
print(f"Squares: {squares}")

# filter() filters elements based on condition
evens = list(filter(lambda x: x % 2 == 0, numbers))
print(f"Even numbers: {evens}")

# sorted() with custom key function
names = ["Alice", "bob", "Charlie", "diana"]
sorted_names = sorted(names, key=lambda name: name.lower())
print(f"Sorted: {sorted_names}")

# Custom callback with reduce-like pattern
def reduce_with_callback(iterable, func, initial, callback):
    """Reduce with callback for each step"""
    accumulator = initial
    for item in iterable:
        accumulator = func(accumulator, item)
        callback(accumulator, item)  # Call callback with current state
    return accumulator

def log_step(acc, item):
    print(f"  Step: {acc} (added {item})")

total = reduce_with_callback(
    [1, 2, 3, 4, 5], 
    lambda acc, x: acc + x, 
    0, 
    log_step
)
print(f"Final total: {total}")

Asyncio Approach (Modern Python)

import asyncio

async def fetch_data(url, callback):
    """Simulate fetching data asynchronously"""
    print(f"Fetching from {url}...")
    await asyncio.sleep(2)  # Simulate network delay
    data = {"url": url, "status": "success", "data": f"Content from {url}"}
    await callback(data)

async def handle_response(data):
    """Async callback to handle response"""
    print(f"Received data: {data}")

async def main():
    """Main async function"""
    # Create multiple fetch operations
    tasks = [
        fetch_data("https://api.example.com/users", handle_response),
        fetch_data("https://api.example.com/posts", handle_response),
    ]
  
    print("Starting async operations...")
    await asyncio.gather(*tasks)
    print("All operations complete!")

# Run async example
# asyncio.run(main())

---

Challenges

1. delayedMessage

Write a function delayed_message(message, delay, callback) that prints the message after the delay, then calls the callback.

💡 Show Answer

import time
import threading

def delayed_message(message, delay_seconds, callback):
    """Print message after delay and call callback"""
    def worker():
        time.sleep(delay_seconds)
        print(message)
        callback()
  
    thread = threading.Thread(target=worker)
    thread.start()

def completion_callback():
    print("Callback executed!")

# Usage
delayed_message("This is the delayed message", 1, completion_callback)
print("Function called, waiting for message...")

# Async version
import asyncio

async def delayed_message_async(message, delay_seconds, callback):
    """Async version of delayed message"""
    await asyncio.sleep(delay_seconds)
    print(message)
    await callback()

async def async_completion_callback():
    print("Async callback executed!")

# Usage (uncomment to run)
# async def test_async():
#     await delayed_message_async("Async delayed message", 1, async_completion_callback)
#     print("Async function complete!")
# 
# asyncio.run(test_async())

---

2. square with callback

Use a callback to calculate the square of a number after a short delay.

💡 Show Answer

import time
import threading
import asyncio

def square_with_callback(number, callback):
    """Calculate square after delay and call callback with result"""
    def worker():
        time.sleep(0.5)  # 500ms delay
        result = number * number
        callback(result)
  
    threading.Thread(target=worker).start()

def print_result(result):
    print(f"Square result: {result}")

# Usage
square_with_callback(5, print_result)
square_with_callback(8, lambda x: print(f"8 squared = {x}"))

# More advanced version with error handling
def safe_square_with_callback(number, success_callback, error_callback=None):
    """Calculate square with error handling"""
    def worker():
        try:
            time.sleep(0.5)
            if not isinstance(number, (int, float)):
                raise TypeError("Number must be int or float")
            result = number * number
            success_callback(result)
        except Exception as e:
            if error_callback:
                error_callback(e)
            else:
                print(f"Error: {e}")
  
    threading.Thread(target=worker).start()

# Usage with error handling
safe_square_with_callback(
    7, 
    lambda x: print(f"Success: {x}"),
    lambda e: print(f"Error occurred: {e}")
)

safe_square_with_callback(
    "invalid",
    lambda x: print(f"Success: {x}"),
    lambda e: print(f"Error occurred: {e}")
)

# Async version
async def square_async(number, callback):
    """Async version of square calculation"""
    await asyncio.sleep(0.5)
    result = number * number
    await callback(result)

async def async_print_result(result):
    print(f"Async square result: {result}")

# Usage (uncomment to run)
# async def test_square_async():
#     await square_async(6, async_print_result)
# 
# asyncio.run(test_square_async())

---

3. getUser

Create a function get_user(callback) that simulates returning a user object after a delay.

💡 Show Answer

import time
import threading
import asyncio
import random

def get_user(callback, user_id=None):
    """Simulate fetching user data after delay"""
    def worker():
        time.sleep(0.5)  # Simulate database/API delay
    
        # Simulate different users
        users = [
            {"id": 1, "name": "Alice", "email": "alice@example.com", "role": "admin"},
            {"id": 2, "name": "Bob", "email": "bob@example.com", "role": "user"},
            {"id": 3, "name": "Charlie", "email": "charlie@example.com", "role": "moderator"},
        ]
    
        if user_id:
            user = next((u for u in users if u["id"] == user_id), None)
        else:
            user = random.choice(users)
    
        callback(user)
  
    threading.Thread(target=worker).start()

def handle_user(user):
    if user:
        print(f"User found: {user['name']} ({user['email']}) - Role: {user['role']}")
    else:
        print("User not found")

# Usage
get_user(handle_user)
get_user(lambda u: print(f"Lambda user: {u}"), user_id=2)

# Advanced version with error handling and multiple callbacks
def get_user_advanced(success_callback, error_callback=None, user_id=None, simulate_error=False):
    """Get user with comprehensive callback handling"""
    def worker():
        try:
            time.sleep(0.5)
        
            if simulate_error:
                raise ConnectionError("Database connection failed")
        
            users_db = {
                1: {"id": 1, "name": "Alice", "email": "alice@example.com", "role": "admin", "active": True},
                2: {"id": 2, "name": "Bob", "email": "bob@example.com", "role": "user", "active": True},
                3: {"id": 3, "name": "Charlie", "email": "charlie@example.com", "role": "moderator", "active": False},
            }
        
            if user_id:
                user = users_db.get(user_id)
                if not user:
                    raise ValueError(f"User with ID {user_id} not found")
            else:
                user = random.choice(list(users_db.values()))
        
            success_callback(user)
        
        except Exception as e:
            if error_callback:
                error_callback(e)
            else:
                print(f"Error getting user: {e}")
  
    threading.Thread(target=worker).start()

def user_success(user):
    print(f"✅ Successfully loaded user: {user['name']}")
    if not user['active']:
        print("⚠️  Warning: User account is inactive")

def user_error(error):
    print(f"❌ Failed to load user: {error}")

# Usage with error handling
print("Testing advanced user loading:")
get_user_advanced(user_success, user_error, user_id=1)
get_user_advanced(user_success, user_error, user_id=999)  # Non-existent user
get_user_advanced(user_success, user_error, simulate_error=True)

# Async version with modern Python patterns
async def get_user_async(callback, user_id=None, simulate_delay=True):
    """Async version of get_user"""
    if simulate_delay:
        await asyncio.sleep(0.5)
  
    users = [
        {"id": 1, "name": "Alice", "email": "alice@example.com", "role": "admin"},
        {"id": 2, "name": "Bob", "email": "bob@example.com", "role": "user"},
        {"id": 3, "name": "Charlie", "email": "charlie@example.com", "role": "moderator"},
    ]
  
    if user_id:
        user = next((u for u in users if u["id"] == user_id), None)
    else:
        user = random.choice(users)
  
    await callback(user)

async def async_handle_user(user):
    """Async callback to handle user"""
    if user:
        print(f"Async user loaded: {user['name']} ({user['role']})")
    else:
        print("Async: User not found")

# Multiple user loading with callbacks
def load_multiple_users(user_ids, callback, complete_callback=None):
    """Load multiple users and call callback for each"""
    loaded_users = []
    remaining = len(user_ids)
  
    def user_loaded(user):
        nonlocal remaining
        loaded_users.append(user)
        callback(user)
        remaining -= 1
    
        if remaining == 0 and complete_callback:
            complete_callback(loaded_users)
  
    for user_id in user_ids:
        get_user_advanced(user_loaded, None, user_id)

def all_users_loaded(users):
    print(f"\n🎉 All users loaded! Total: {len(users)}")
    for user in users:
        if user:  # Filter out None values
            print(f"  - {user['name']} ({user['role']})")

# Load multiple users
print("\nLoading multiple users:")
load_multiple_users(
    [1, 2, 3], 
    lambda u: print(f"  Loaded: {u['name'] if u else 'Not found'}"),
    all_users_loaded
)

# Usage (uncomment to run async version)
# async def test_async_users():
#     print("Testing async user loading:")
#     await get_user_async(async_handle_user, user_id=1)
#     await get_user_async(async_handle_user, user_id=2)
# 
# asyncio.run(test_async_users())

---

4. processItems

Create a function that processes an array of items using callbacks for each step.

💡 Show Answer

import time
import threading
from typing import List, Callable, Any
import asyncio

def process_items(items: List[Any], process_callback: Callable, complete_callback: Callable = None):
    """Process array of items with callbacks"""
    results = []
    processed_count = 0
    total_items = len(items)
  
    def item_processed(original_item, result):
        nonlocal processed_count
        results.append({"original": original_item, "result": result})
        processed_count += 1
    
        print(f"Processed {processed_count}/{total_items}: {original_item} -> {result}")
    
        if processed_count == total_items and complete_callback:
            complete_callback(results)
  
    def worker():
        for item in items:
            time.sleep(0.2)  # Simulate processing time
            try:
                result = process_callback(item)
                item_processed(item, result)
            except Exception as e:
                item_processed(item, f"Error: {e}")
  
    threading.Thread(target=worker).start()

# Usage
def double_number(x):
    if isinstance(x, (int, float)):
        return x * 2
    raise TypeError("Must be a number")

def all_items_processed(results):
    print(f"\n✅ All {len(results)} items processed!")
    successful = [r for r in results if not str(r['result']).startswith('Error')]
    print(f"Successful: {len(successful)}, Failed: {len(results) - len(successful)}")

numbers = [1, 2, 3, 4, 5]
process_items(numbers, double_number, all_items_processed)

# Advanced version with different processing strategies
def process_items_advanced(
    items: List[Any], 
    process_callback: Callable,
    progress_callback: Callable = None,
    complete_callback: Callable = None,
    error_callback: Callable = None,
    parallel: bool = False
):
    """Advanced item processing with multiple callback types"""
    results = []
    errors = []
  
    def process_single_item(item, index):
        try:
            time.sleep(0.1)  # Simulate processing
            result = process_callback(item)
        
            if progress_callback:
                progress_callback(index + 1, len(items), item, result)
        
            return {"index": index, "item": item, "result": result, "success": True}
        except Exception as e:
            error_info = {"index": index, "item": item, "error": str(e), "success": False}
            if error_callback:
                error_callback(error_info)
            return error_info
  
    def worker():
        nonlocal results, errors
    
        if parallel:
            # Process items in parallel using threads
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
                futures = [executor.submit(process_single_item, item, i) for i, item in enumerate(items)]
                for future in concurrent.futures.as_completed(futures):
                    result = future.result()
                    if result["success"]:
                        results.append(result)
                    else:
                        errors.append(result)
        else:
            # Process items sequentially
            for i, item in enumerate(items):
                result = process_single_item(item, i)
                if result["success"]:
                    results.append(result)
                else:
                    errors.append(result)
    
        if complete_callback:
            complete_callback(results, errors)
  
    threading.Thread(target=worker).start()

# Example usage
def complex_processor(item):
    """Complex processing function"""
    if isinstance(item, str):
        return item.upper()
    elif isinstance(item, (int, float)):
        return item ** 2
    elif isinstance(item, list):
        return len(item)
    else:
        raise TypeError(f"Cannot process type {type(item)}")

def progress_update(current, total, item, result):
    percentage = (current / total) * 100
    print(f"Progress: {percentage:.1f}% - {item} -> {result}")

def error_handler(error_info):
    print(f"❌ Error processing item {error_info['index']}: {error_info['error']}")

def processing_complete(results, errors):
    print(f"\n🎉 Processing complete!")
    print(f"✅ Successful: {len(results)}")
    print(f"❌ Errors: {len(errors)}")
    if results:
        print("Results:")
        for r in results:
            print(f"  {r['item']} -> {r['result']}")

# Test with mixed data types
mixed_items = ["hello", 42, [1, 2, 3], "world", 3.14, {"invalid": "dict"}]

print("Processing mixed items (sequential):")
process_items_advanced(
    mixed_items,
    complex_processor,
    progress_update,
    processing_complete,
    error_handler,
    parallel=False
)

time.sleep(2)  # Wait for first batch to complete

print("\nProcessing mixed items (parallel):")
process_items_advanced(
    mixed_items,
    complex_processor,
    progress_update,
    processing_complete,
    error_handler,
    parallel=True
)

# Async version for modern Python
async def process_items_async(items, process_callback, progress_callback=None, complete_callback=None):
    """Async version of item processing"""
    results = []
  
    async def process_item(item, index):
        await asyncio.sleep(0.1)  # Simulate async processing
        result = await process_callback(item) if asyncio.iscoroutinefunction(process_callback) else process_callback(item)
    
        if progress_callback:
            await progress_callback(index + 1, len(items), item, result)
    
        return {"item": item, "result": result}
  
    # Process all items concurrently
    tasks = [process_item(item, i) for i, item in enumerate(items)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
  
    if complete_callback:
        await complete_callback(results)

async def async_progress(current, total, item, result):
    percentage = (current / total) * 100
    print(f"Async Progress: {percentage:.1f}% - {item} -> {result}")

async def async_complete(results):
    print(f"Async processing complete! Processed {len(results)} items")

# Usage (uncomment to run)
# async def test_async_processing():
#     items = [1, 2, 3, 4, 5]
#     await process_items_async(
#         items,
#         lambda x: x * 3,
#         async_progress,
#         async_complete
#     )
# 
# asyncio.run(test_async_processing())

---

5. Event System

Create a simple event system using callbacks for event handling.

💡 Show Answer

import time
import threading
from typing import Dict, List, Callable, Any
from dataclasses import dataclass
from datetime import datetime

@dataclass
class Event:
    name: str
    data: Any
    timestamp: datetime

class EventEmitter:
    """Simple event system using callbacks"""
  
    def __init__(self):
        self._listeners: Dict[str, List[Callable]] = {}
        self._once_listeners: Dict[str, List[Callable]] = {}
        self._event_history: List[Event] = []
  
    def on(self, event_name: str, callback: Callable):
        """Register a persistent event listener"""
        if event_name not in self._listeners:
            self._listeners[event_name] = []
        self._listeners[event_name].append(callback)
  
    def once(self, event_name: str, callback: Callable):
        """Register a one-time event listener"""
        if event_name not in self._once_listeners:
            self._once_listeners[event_name] = []
        self._once_listeners[event_name].append(callback)
  
    def off(self, event_name: str, callback: Callable = None):
        """Remove event listener(s)"""
        if callback is None:
            # Remove all listeners for this event
            self._listeners.pop(event_name, None)
            self._once_listeners.pop(event_name, None)
        else:
            # Remove specific callback
            if event_name in self._listeners:
                try:
                    self._listeners[event_name].remove(callback)
                except ValueError:
                    pass
            if event_name in self._once_listeners:
                try:
                    self._once_listeners[event_name].remove(callback)
                except ValueError:
                    pass
  
    def emit(self, event_name: str, data: Any = None):
        """Emit an event to all listeners"""
        event = Event(event_name, data, datetime.now())
        self._event_history.append(event)
    
        # Call persistent listeners
        if event_name in self._listeners:
            for callback in self._listeners[event_name][:]:  # Copy to avoid modification during iteration
                try:
                    callback(event)
                except Exception as e:
                    print(f"Error in event listener for '{event_name}': {e}")
    
        # Call once listeners and remove them
        if event_name in self._once_listeners:
            once_callbacks = self._once_listeners.pop(event_name, [])
            for callback in once_callbacks:
                try:
                    callback(event)
                except Exception as e:
                    print(f"Error in once listener for '{event_name}': {e}")
  
    def emit_async(self, event_name: str, data: Any = None, delay: float = 0):
        """Emit an event asynchronously after optional delay"""
        def worker():
            if delay > 0:
                time.sleep(delay)
            self.emit(event_name, data)
    
        threading.Thread(target=worker).start()
  
    def get_listener_count(self, event_name: str) -> int:
        """Get number of listeners for an event"""
        persistent = len(self._listeners.get(event_name, []))
        once = len(self._once_listeners.get(event_name, []))
        return persistent + once
  
    def get_event_history(self, event_name: str = None, limit: int = None) -> List[Event]:
        """Get event history"""
        events = self._event_history
        if event_name:
            events = [e for e in events if e.name == event_name]
        if limit:
            events = events[-limit:]
        return events

# Example Usage
emitter = EventEmitter()

# Define event handlers
def user_login_handler(event):
    print(f"🔑 User logged in: {event.data['username']} at {event.timestamp.strftime('%H:%M:%S')}")

def user_logout_handler(event):
    print(f"👋 User logged out: {event.data['username']}")

def security_monitor(event):
    if event.name in ['user_login', 'user_logout', 'failed_login']:
        print(f"🛡️  Security log: {event.name} - {event.data}")

def welcome_message(event):
    print(f"🎉 Welcome to the system, {event.data['username']}!")

# Register event listeners
emitter.on('user_login', user_login_handler)
emitter.on('user_logout', user_logout_handler)
emitter.on('user_login', security_monitor)
emitter.on('user_logout', security_monitor)
emitter.on('failed_login', security_monitor)

# One-time welcome message
emitter.once('user_login', welcome_message)

# Simulate events
print("Simulating user events...")
emitter.emit('user_login', {'username': 'alice', 'ip': '192.168.1.100'})
emitter.emit('user_login', {'username': 'bob', 'ip': '192.168.1.101'})  # Welcome won't fire again
emitter.emit('failed_login', {'username': 'hacker', 'ip': '10.0.0.1'})
emitter.emit('user_logout', {'username': 'alice'})

print(f"\nListener counts:")
print(f"user_login: {emitter.get_listener_count('user_login')}")
print(f"user_logout: {emitter.get_listener_count('user_logout')}")

# Advanced example: File watcher simulation
class FileWatcher(EventEmitter):
    """File watcher that emits events for file changes"""
  
    def __init__(self):
        super().__init__()
        self._watching = {}
        self._running = False
  
    def watch_file(self, filepath: str):
        """Start watching a file"""
        self._watching[filepath] = {
            'last_modified': time.time(),
            'size': 0
        }
    
        if not self._running:
            self._start_monitoring()
  
    def _start_monitoring(self):
        """Start the monitoring loop"""
        self._running = True
    
        def monitor():
            while self._running:
                time.sleep(1)  # Check every second
                for filepath in list(self._watching.keys()):
                    self._check_file_changes(filepath)
    
        threading.Thread(target=monitor, daemon=True).start()
  
    def _check_file_changes(self, filepath: str):
        """Check if file has changed"""
        try:
            import os
            if os.path.exists(filepath):
                current_time = os.path.getmtime(filepath)
                current_size = os.path.getsize(filepath)
            
                last_modified = self._watching[filepath]['last_modified']
                last_size = self._watching[filepath]['size']
            
                if current_time > last_modified:
                    self.emit('file_modified', {
                        'filepath': filepath,
                        'old_size': last_size,
                        'new_size': current_size
                    })
                
                    self._watching[filepath]['last_modified'] = current_time
                    self._watching[filepath]['size'] = current_size
            else:
                self.emit('file_deleted', {'filepath': filepath})
                del self._watching[filepath]
        except Exception as e:
            self.emit('watch_error', {'filepath': filepath, 'error': str(e)})
  
    def stop_watching(self):
        """Stop file monitoring"""
        self._running = False
        self._watching.clear()

# File watcher example
file_watcher = FileWatcher()

def file_change_handler(event):
    print(f"📁 File changed: {event.data['filepath']}")
    print(f"   Size: {event.data['old_size']} -> {event.data['new_size']} bytes")

def file_deleted_handler(event):
    print(f"🗑️  File deleted: {event.data['filepath']}")

def file_error_handler(event):
    print(f"❌ Watch error: {event.data['error']}")

file_watcher.on('file_modified', file_change_handler)
file_watcher.on('file_deleted', file_deleted_handler)
file_watcher.on('watch_error', file_error_handler)

# Start watching (uncomment to test with real files)
# file_watcher.watch_file('/path/to/your/file.txt')

print("\n📊 Event History:")
for event in emitter.get_event_history(limit=5):
    print(f"  {event.timestamp.strftime('%H:%M:%S')} - {event.name}: {event.data}")