Skip to content

Function Composition & Piping

Overview

Function composition combines simple functions into complex workflows:

  • Composition: f(g(x)) - apply functions in sequence
  • Piping: x|> g|> f - data-first style
  • Builder pattern: Chain methods fluently
  • Workflow automation: Clear data transformations

-

Function Composition

Basic Composition

# Simple functions
double = lambda x: x * 2
add_one = lambda x: x + 1
square = lambda x: x ** 2

# Compose manually
result = square(add_one(double(5))) # ((5*2)+1)² = 121

# Composition utility
def compose(*functions):
 """Compose functions (right to left)."""
 def composed(x):
 result = x
 for func in reversed(functions):
 result = func(result)
 return result
 return composed

# Now compose elegantly
pipeline = compose(square, add_one, double)
print(pipeline(5)) # 121

# Order matters!
pipeline2 = compose(double, add_one, square)
print(pipeline2(5)) # (5²+1)*2 = 52

Piping Style

# Pipe function (left-to-right, opposite of compose)
def pipe(*functions):
 """Apply functions left to right."""
 def piped(x):
 result = x
 for func in functions:
 result = func(result)
 return result
 return piped

# More intuitive for data flow
normalize = lambda x: (x - x.mean()) / x.std()
scale = lambda x: x * 255
quantize = lambda x: x.astype(int)

preprocess = pipe(normalize, scale, quantize)
# Data flows

Composition Operators

Operator Overloading for Composition

class Composable:
 """Function wrapper that supports composition."""

 def __init__(self, func):
 self.func = func

 def __call__(self, *args, **kwargs):
 return self.func(*args, **kwargs)

 def __rshift__(self, other):
 """f >> g means g(f(x))."""
 return Composable(lambda x: other(self.func(x)))

 def __lshift__(self, other):
 """f << g means f(g(x))."""
 return Composable(lambda x: self.func(other(x)))

# Use it
square = Composable(lambda x: x ** 2)
add_one = Composable(lambda x: x + 1)
double = Composable(lambda x: x * 2)

# Compose with >> operator
pipeline = double >> add_one >> square
print(pipeline(5)) # ((5*2)+1)² = 121

-

Data Transformation Pipelines

ML Data Pipeline

import numpy as np

def create_pipeline(*steps):
 """Create data transformation pipeline."""
 def process(data):
 result = data
 for step in steps:
 result = step(result)
 return result
 return process

# Define transformations
def load_data(filepath):
 """Load numpy array."""
 return np.random.randn(100, 10) # Placeholder

def remove_outliers(data):
 """Remove outliers (3-sigma)."""
 mean = data.mean(axis=0)
 std = data.std(axis=0)
 mask = np.abs(data - mean) < 3 * std
 return data[mask.all(axis=1)]

def normalize(data):
 """Normalize to mean=0, std=1."""
 return (data - data.mean(axis=0)) / data.std(axis=0)

def pca_reduce(data, components=5):
 """Reduce with PCA."""
 # Simplified PCA
 return data[:,:components]

# Create pipeline
data_pipeline = create_pipeline(
 load_data,
 remove_outliers,
 normalize,
 lambda x: pca_reduce(x, 5)
)

# Use it
processed_data = data_pipeline("path/to/data.csv")
print(processed_data.shape) # (100, 5)

Inference Pipeline

# Model inference pipeline
class Model:
 def forward(self, x):
 # Simplified model
 return x * 2

def create_inference_pipeline(model):
 """Create model inference pipeline."""

 def preprocess(raw_input):
 """Convert raw input to tensor."""
 return raw_input

 def inference(preprocessed):
 """Run model."""
 return model.forward(preprocessed)

 def postprocess(output):
 """Convert output to result."""
 return {"prediction": output, "confidence": 0.95}

 # Compose into pipeline
 return pipe(preprocess, inference, postprocess)

model = Model()
pipeline = create_inference_pipeline(model)

# Use end-to-end
result = pipeline(5)
print(result) # {'prediction': 10, 'confidence': 0.95}

-

Builder Pattern

Fluent Interface

class DataProcessing:
 """Fluent data processing builder."""

 def __init__(self, data):
 self.data = data

 def normalize(self, mean=None, std=None):
 """Normalize data."""
 data = self.data
 mean = mean or data.mean(axis=0)
 std = std or data.std(axis=0)
 self.data = (data - mean) / std
 return self # Return self for chaining

 def scale(self, factor):
 """Scale data."""
 self.data = self.data * factor
 return self

 def filter_outliers(self, std_threshold=3):
 """Remove outliers."""
 data = self.data
 mask = np.abs(data) < std_threshold * np.std(data)
 self.data = data[mask.all(axis=1)]
 return self

 def get(self):
 """Get processed data."""
 return self.data

# Use fluent interface
import numpy as np

data = np.random.randn(100, 10)
processed = (DataProcessing(data)
.normalize()
.scale(255)
.filter_outliers(3)
.get())

print(processed.shape) # Transformed data shape

-

Advanced Composition

Monadic Composition

# Composition with error handling
class Result:
 """Represents success or failure."""

 def __init__(self, value=None, error=None):
 self.value = value
 self.error = error

 def is_success(self):
 return self.error is None

 def flat_map(self, func):
 """Apply function if successful."""
 if self.is_success():
 try:
 return func(self.value)
 except Exception as e:
 return Result(error=str(e))
 return self

 def map(self, func):
 """Transform value if successful."""
 return self.flat_map(lambda x: Result(func(x)))

 def get_or_default(self, default):
 """Get value or default if failed."""
 return self.value if self.is_success() else default

# Use monadic composition
def safe_divide(x, y):
 if y == 0:
 return Result(error="Division by zero")
 return Result(x / y)

def safe_sqrt(x):
 if x < 0:
 return Result(error="Negative number")
 return Result(x ** 0.5)

# Compose operations that can fail
result = (Result(16)
.flat_map(lambda x: safe_sqrt(x)) # sqrt(16) = 4
.flat_map(lambda x: safe_divide(8, x)) # 8/4 = 2
.get_or_default(0))

print(result) # 2.0

Real-World ML Pipelines

Training Pipeline

class TrainingPipeline:
 """Chainable training pipeline."""

 def __init__(self, model):
 self.model = model
 self.optimizer = None
 self.loss_fn = None
 self.metrics = []

 def with_optimizer(self, optimizer):
 self.optimizer = optimizer
 return self

 def with_loss(self, loss_fn):
 self.loss_fn = loss_fn
 return self

 def track_metric(self, metric_name):
 self.metrics.append(metric_name)
 return self

 def compile(self):
 """Verify all components are set."""
 if not self.optimizer:
 raise ValueError("Optimizer required")
 if not self.loss_fn:
 raise ValueError("Loss function required")
 return self

 def train(self, data, epochs):
 """Run training."""
 print(f"Training {self.model} for {epochs} epochs")
 print(f"Optimizer: {self.optimizer}")
 print(f"Loss: {self.loss_fn}")
 print(f"Tracking: {self.metrics}")
 # Actual training code here

# Use it
import torch
model = torch.nn.Linear(10, 5)

(TrainingPipeline(model)
.with_optimizer('adam')
.with_loss('mse')
.track_metric('accuracy')
.track_metric('loss')
.compile()
.train(data, epochs=10))

Summary: Composition Patterns

Pattern Use Case Example
Function composition Chain operations compose(f, g, h)
Piping Left-to-right data flow pipe(f, g, h)(x)
Builder Fluent configuration obj.set_a().set_b().build()
Monadic Handle errors in chain result.flat_map(f)
Pipeline Multi-step workflow preprocess → model → postprocess

-

  • 01 Functional Programming Fundamentals - Pure functions for composition
  • [02 Higher Order Functions & Closures](/05-py3/03-functional-programming/(02-higher-order-functions-closures/) - Functions that compose
  • [03 Iterator & Generator Protocol](/05-py3/01-fundamentals/(03-iterator-generator-protocol/) - Lazy composition
  • 00 Readme - Performance of composed functions