Advanced Applications of Higher-Order Functions

Let's explore some more advanced applications of higher-order functions in real-world scenarios.

Domain-Specific Languages (DSLs)

# File: simple_dsl.py
# Location: /python_projects/functions_tutorial/

# A simple DSL for query building
def select(*columns):
    """Start building a query by selecting columns."""
    query = {"columns": columns, "filters": [], "order_by": None, "limit": None}
    
    def from_table(table):
        """Specify the table to query."""
        query["table"] = table
        
        # Return an object with methods for further query building
        return QueryBuilder(query)
    
    return from_table

class QueryBuilder:
    """A query builder that uses method chaining."""
    
    def __init__(self, query):
        self.query = query
    
    def where(self, column, operator, value):
        """Add a filter condition."""
        self.query["filters"].append((column, operator, value))
        return self
    
    def order_by(self, column, direction="ASC"):
        """Set the order by clause."""
        self.query["order_by"] = (column, direction)
        return self
    
    def limit(self, count):
        """Set the limit clause."""
        self.query["limit"] = count
        return self
    
    def build(self):
        """Build the final SQL query string."""
        columns = ", ".join(self.query["columns"])
        sql = f"SELECT {columns} FROM {self.query['table']}"
        
        if self.query["filters"]:
            conditions = []
            for column, operator, value in self.query["filters"]:
                if isinstance(value, str):
                    conditions.append(f"{column} {operator} '{value}'")
                else:
                    conditions.append(f"{column} {operator} {value}")
            
            sql += " WHERE " + " AND ".join(conditions)
        
        if self.query["order_by"]:
            column, direction = self.query["order_by"]
            sql += f" ORDER BY {column} {direction}"
        
        if self.query["limit"] is not None:
            sql += f" LIMIT {self.query['limit']}"
        
        return sql

# Using the DSL
query1 = (select("id", "name", "email")
    .from_table("users")
    .where("age", ">", 18)
    .where("status", "=", "active")
    .order_by("name")
    .limit(10)
    .build())

print(f"Query 1: {query1}")

query2 = (select("product_id", "name", "price")
    .from_table("products")
    .where("category", "=", "Electronics")
    .where("price", "<", 1000)
    .order_by("price", "DESC")
    .build())

print(f"Query 2: {query2}")

# A different kind of DSL using higher-order functions for data validation
def is_required(field_name):
    """Create a validator that checks if a field is present and not empty."""
    def validator(data):
        value = data.get(field_name)
        if value is None or (isinstance(value, str) and not value.strip()):
            return False, f"{field_name} is required"
        return True, None
    return validator

def min_length(field_name, min_len):
    """Create a validator that checks if a field meets a minimum length."""
    def validator(data):
        value = data.get(field_name)
        if value is None or len(value) < min_len:
            return False, f"{field_name} must be at least {min_len} characters"
        return True, None
    return validator

def is_email(field_name):
    """Create a validator that checks if a field is a valid email."""
    def validator(data):
        value = data.get(field_name)
        if value is None or '@' not in value or '.' not in value:
            return False, f"{field_name} must be a valid email"
        return True, None
    return validator

def validate(data, *validators):
    """Validate data against a series of validators."""
    errors = []
    
    for validator in validators:
        is_valid, error = validator(data)
        if not is_valid:
            errors.append(error)
    
    return len(errors) == 0, errors

# Using the validation DSL
user_data = {
    "username": "jdoe",
    "email": "john@example.com",
    "password": "pass"
}

is_valid, errors = validate(
    user_data,
    is_required("username"),
    is_required("email"),
    is_required("password"),
    min_length("username", 3),
    min_length("password", 8),
    is_email("email")
)

if is_valid:
    print("User data is valid")
else:
    print("Validation errors:")
    for error in errors:
        print(f"  - {error}")

Domain-Specific Languages (DSLs) are specialized languages for particular domains or problems. Higher-order functions can be used to create expressive, fluent APIs that serve as internal DSLs in Python.

In the example above, we created two simple DSLs:

• A SQL query builder that allows chaining methods to construct queries
• A data validation system that uses higher-order functions to create reusable validators

These DSLs make the code more readable and expressive, closer to the domain language, and less prone to errors.

Reactive Programming

# File: reactive_example.py
# Location: /python_projects/functions_tutorial/

class Observable:
    """A simple Observable implementation for reactive programming."""
    
    def __init__(self, initial_value=None):
        self.value = initial_value
        self.observers = []
    
    def subscribe(self, observer):
        """Add an observer function that will be called when the value changes."""
        self.observers.append(observer)
        # Call the observer with the current value
        observer(self.value)
        return self  # For method chaining
    
    def unsubscribe(self, observer):
        """Remove an observer function."""
        if observer in self.observers:
            self.observers.remove(observer)
        return self  # For method chaining
    
    def set(self, new_value):
        """Set a new value and notify all observers."""
        if new_value != self.value:
            self.value = new_value
            for observer in self.observers:
                observer(new_value)
        return self  # For method chaining
    
    def map(self, transform_function):
        """
        Create a new Observable that transforms the values of this Observable.
        
        Args:
            transform_function: A function to apply to each value
            
        Returns:
            A new Observable with the transformed values
        """
        result = Observable()
        
        def update(value):
            result.set(transform_function(value))
        
        self.subscribe(update)
        return result
    
    def filter(self, predicate_function):
        """
        Create a new Observable that only emits values that satisfy the predicate.
        
        Args:
            predicate_function: A function that returns True for values to emit
            
        Returns:
            A new Observable with the filtered values
        """
        result = Observable()
        
        def update(value):
            if predicate_function(value):
                result.set(value)
        
        self.subscribe(update)
        return result
    
    def combine(self, other_observable, combiner_function):
        """
        Combine this Observable with another using a combiner function.
        
        Args:
            other_observable: Another Observable to combine with
            combiner_function: A function that takes two values and returns a combined value
            
        Returns:
            A new Observable with the combined values
        """
        result = Observable()
        latest_values = [self.value, other_observable.value]
        
        def update_first(value):
            latest_values[0] = value
            result.set(combiner_function(latest_values[0], latest_values[1]))
        
        def update_second(value):
            latest_values[1] = value
            result.set(combiner_function(latest_values[0], latest_values[1]))
        
        self.subscribe(update_first)
        other_observable.subscribe(update_second)
        
        return result

# Using the reactive programming system
def create_temperature_converter():
    """Create a temperature converter with reactive bindings."""
    # Observable temperature values
    celsius = Observable(0)
    fahrenheit = Observable(32)
    
    # Create bidirectional binding
    c_to_f_binding = celsius.map(lambda c: c * 9/5 + 32)
    f_to_c_binding = fahrenheit.map(lambda f: (f - 32) * 5/9)
    
    # Subscribe to updates
    c_to_f_binding.subscribe(lambda f: fahrenheit.set(f))
    f_to_c_binding.subscribe(lambda c: celsius.set(c))
    
    return celsius, fahrenheit

# Create the converter
celsius, fahrenheit = create_temperature_converter()

# Add observers for display
celsius.subscribe(lambda c: print(f"Celsius: {c:.2f}°C"))
fahrenheit.subscribe(lambda f: print(f"Fahrenheit: {f:.2f}°F"))

print("\nSetting Celsius to 25°C:")
celsius.set(25)

print("\nSetting Fahrenheit to 68°F:")
fahrenheit.set(68)

# More advanced example with multiple observables
def create_weather_monitor():
    """Create a weather monitoring system with derived values."""
    temperature = Observable(20)  # Celsius
    humidity = Observable(50)     # Percent
    wind_speed = Observable(10)   # km/h
    
    # Derived values
    feels_like = temperature.combine(
        wind_speed,
        lambda temp, wind: temp - wind * 0.1  # Simplified wind chill calculation
    )
    
    comfort_level = temperature.combine(
        humidity,
        lambda temp, humid: (
            "Comfortable" if 18 <= temp <= 24 and 40 <= humid <= 60
            else "Uncomfortable"
        )
    )
    
    # High temperature warning
    temperature.filter(lambda temp: temp > 30).subscribe(
        lambda temp: print(f"WARNING: High temperature detected: {temp}°C")
    )
    
    return {
        "temperature": temperature,
        "humidity": humidity,
        "wind_speed": wind_speed,
        "feels_like": feels_like,
        "comfort_level": comfort_level
    }

# Create the weather monitor
weather = create_weather_monitor()

# Add observers
weather["feels_like"].subscribe(lambda temp: print(f"Feels like: {temp:.1f}°C"))
weather["comfort_level"].subscribe(lambda level: print(f"Comfort level: {level}"))

print("\nUpdating temperature to 22°C:")
weather["temperature"].set(22)

print("\nIncreasing humidity to 70%:")
weather["humidity"].set(70)

print("\nSetting temperature to 35°C (should trigger warning):")
weather["temperature"].set(35)

Reactive programming is a paradigm focused on data flows and the propagation of changes. Higher-order functions are fundamental to this paradigm, allowing the creation of data streams that can be transformed, filtered, and combined.

In the example above, we implemented a simple reactive system with Observable objects that can be transformed and combined using higher-order functions. This approach is useful for:

• User interfaces with complex data dependencies
• Real-time data processing
• Event-driven systems
• Bidirectional data binding

Libraries like RxPy (Reactive Extensions for Python) provide more sophisticated implementations of these concepts for real-world applications.

Aspect-Oriented Programming

# File: aspect_oriented.py
# Location: /python_projects/functions_tutorial/

import functools
import time
import threading
import traceback

# Define aspects as higher-order functions

def log_calls(logger=print):
    """
    Aspect that logs function calls with arguments and return values.
    
    Args:
        logger: Function to use for logging (default: print)
        
    Returns:
        A decorator function
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            logger(f"Calling {func.__name__} with args={args}, kwargs={kwargs}")
            result = func(*args, **kwargs)
            logger(f"{func.__name__} returned {result}")
            return result
        return wrapper
    return decorator

def measure_time(logger=print):
    """
    Aspect that measures and logs function execution time.
    
    Args:
        logger: Function to use for logging (default: print)
        
    Returns:
        A decorator function
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            execution_time = end_time - start_time
            logger(f"{func.__name__} took {execution_time:.6f} seconds")
            return result
        return wrapper
    return decorator

def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
    """
    Aspect that retries a function if it raises specified exceptions.
    
    Args:
        max_attempts: Maximum number of attempts
        delay: Delay between attempts in seconds
        exceptions: Tuple of exceptions to catch and retry
        
    Returns:
        A decorator function
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    attempts += 1
                    if attempts == max_attempts:
                        raise
                    print(f"Attempt {attempts} failed with error: {e}")
                    print(f"Retrying in {delay} seconds...")
                    time.sleep(delay)
        return wrapper
    return decorator

def synchronized(lock=None):
    """
    Aspect that ensures a function is thread-safe using a lock.
    
    Args:
        lock: Lock to use (default: create a new lock)
        
    Returns:
        A decorator function
    """
    def decorator(func):
        nonlocal lock
        if lock is None:
            lock = threading.RLock()
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with lock:
                return func(*args, **kwargs)
        return wrapper
    return decorator

# Demonstration of aspect-oriented programming

# A simple function to demonstrate aspects
def process_data(data):
    """Process some data (dummy implementation)."""
    time.sleep(0.1)  # Simulate processing
    return data[::-1]  # Reverse the data

# Apply multiple aspects to the function
@log_calls()
@measure_time()
@retry(max_attempts=3, delay=0.5, exceptions=(ValueError, RuntimeError))
@synchronized()
def enhanced_process_data(data):
    """Process data with enhanced aspects."""
    if not data:
        raise ValueError("Empty data")
    
    time.sleep(0.1)  # Simulate processing
    return data[::-1]  # Reverse the data

# Test the enhanced function
try:
    result1 = enhanced_process_data("Hello, world!")
    print(f"Result: {result1}")
    
    # This should trigger a retry and then fail
    result2 = enhanced_process_data("")
    print(f"Result: {result2}")
except Exception as e:
    print(f"Final error: {e}")

# Custom aspect for validation
def validate_args(**validators):
    """
    Aspect that validates function arguments.
    
    Args:
        validators: A dictionary mapping parameter names to validation functions
        
    Returns:
        A decorator function
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Get function signature to map positional args to parameter names
            import inspect
            sig = inspect.signature(func)
            bound_args = sig.bind(*args, **kwargs)
            bound_args.apply_defaults()
            
            # Validate arguments
            for param_name, validator in validators.items():
                if param_name in bound_args.arguments:
                    value = bound_args.arguments[param_name]
                    if not validator(value):
                        raise ValueError(f"Invalid value for {param_name}: {value}")
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Define validation functions
def is_positive(x):
    return x > 0

def is_string(s):
    return isinstance(s, str)

def min_length(min_len):
    return lambda s: len(s) >= min_len

# Apply validation aspect
@validate_args(
    name=is_string,
    age=is_positive,
    address=lambda s: is_string(s) and len(s) >= 5
)
def create_user(name, age, address):
    """Create a user with validated arguments."""
    return {"name": name, "age": age, "address": address}

# Test validation
try:
    user1 = create_user("Alice", 30, "123 Main St")
    print(f"Valid user: {user1}")
    
    user2 = create_user("Bob", -5, "Home")
    print(f"This shouldn't print: {user2}")
except ValueError as e:
    print(f"Validation error: {e}")

Aspect-Oriented Programming (AOP) is a paradigm that aims to increase modularity by separating cross-cutting concerns. Higher-order functions, especially decorators, are an excellent way to implement aspects in Python.

In the example above, we implemented several aspects as decorators:

• log_calls: Logs function calls with arguments and return values
• measure_time: Measures and logs function execution time
• retry: Retries a function if it raises specified exceptions
• synchronized: Ensures a function is thread-safe using a lock
• validate_args: Validates function arguments against specified validators

These aspects can be applied to any function without modifying its core logic, promoting separation of concerns and code reuse.