Pattern-Driven Condition Design - Thinking Architecturally About Branching Logic
Reading time: ~26 minutes | Level: Foundation → Engineering
Study this code for thirty seconds. It evaluates whether a user can perform an action on a resource:
def can_user_act(user, resource, action):
if user.role == "admin":
return True
if user.role == "owner" and resource.owner_id == user.id:
return action in ("read", "write", "delete")
if user.role == "member" and resource.is_public:
return action == "read"
if user.role == "guest":
return False
return False
It works. But add a new role, a new action type, or a new resource visibility level, and you are rewriting the function. Make the same change in six different places across the codebase, and you have an authorization bug waiting to happen. The root problem is not the code itself - it is the design. This function embeds the authorization rules directly in executable logic. Every rule change is a code change. This lesson teaches you to design condition systems that are open to extension without modification, where adding a new rule means adding data, not editing logic.
What You Will Learn
- The evolution from ad-hoc
if/elifchains to intentional design patterns for branching logic - Pattern 1: Strategy - callable dispatch dictionaries that replace long
if/elifchains - Pattern 2: Rule Engine - ordered lists of
(condition_fn, action_fn)tuples that are fully extensible - Pattern 3: State Machine - states as enums, transitions as dictionaries, making state logic explicit and testable
- Pattern 4: Chain of Responsibility - handlers tried in sequence, first match wins
- Pattern 5: Null Object - replacing scattered
if x is not None:with a do-nothing default object - How to treat condition functions as first-class objects: storing, passing, and composing predicates
- Data-driven conditions: reading rules from config files or databases at runtime
- How each pattern improves testability
- A complete production example: an authorization system built on the rule engine pattern
- When not to use patterns - the cost of abstraction
Prerequisites
- Comfortable with Python functions, including passing functions as arguments
- Understand dictionaries, lists of tuples, and basic class definitions
- Familiarity with the guard clause and dispatch dictionary techniques from the previous lesson
From Ad-Hoc to Intentional Design
Most branching logic starts as a single if. Then a second case is added, then a third. After six months, a simple function is a 40-line if/elif chain that nobody wants to touch:
# Six months of growth, zero design
def calculate_discount(user, cart):
if user.loyalty_tier == "gold":
discount = 0.20
elif user.loyalty_tier == "silver":
discount = 0.10
elif user.loyalty_tier == "bronze":
discount = 0.05
elif cart.total > 500:
discount = 0.08
elif cart.total > 200:
discount = 0.04
elif user.is_first_purchase:
discount = 0.15
elif user.has_promo_code:
discount = cart.promo_discount
else:
discount = 0.0
return discount
The problem is not length - it is brittleness. Adding a new discount rule means modifying this function. Changing rule priority means reordering lines and re-testing the full chain. A/B testing two pricing strategies means branching this function again. The design does not accommodate change gracefully.
Pattern-driven design asks a different question at the outset: what is the structure of this decision system? Once that structure is identified, you choose a pattern that makes the structure explicit, separates the rules from the evaluation machinery, and allows extension without modification.
Pattern 1: Strategy - Callable Dispatch
The problem it solves: A long if/elif chain where each branch does the same kind of operation but differently, driven by the value of a single discriminating variable.
The mechanism: A dictionary maps discriminating values to callable functions (the "strategies"). The evaluation machinery is a single dictionary lookup followed by a call.
# Before: if/elif chain
def process_payment(method, amount, currency):
if method == "stripe":
return stripe_api.charge(amount, currency)
elif method == "paypal":
return paypal_api.execute(amount, currency)
elif method == "bank_transfer":
return bank_api.initiate_transfer(amount, currency)
elif method == "crypto":
return crypto_api.send(amount, currency)
else:
raise ValueError(f"Unknown payment method: {method}")
# After: Strategy pattern via callable dispatch
def stripe_strategy(amount, currency):
return stripe_api.charge(amount, currency)
def paypal_strategy(amount, currency):
return paypal_api.execute(amount, currency)
def bank_transfer_strategy(amount, currency):
return bank_api.initiate_transfer(amount, currency)
def crypto_strategy(amount, currency):
return crypto_api.send(amount, currency)
PAYMENT_STRATEGIES = {
"stripe": stripe_strategy,
"paypal": paypal_strategy,
"bank_transfer": bank_transfer_strategy,
"crypto": crypto_strategy,
}
def process_payment(method, amount, currency):
strategy = PAYMENT_STRATEGIES.get(method)
if strategy is None:
raise ValueError(f"Unknown payment method: {method}")
return strategy(amount, currency)
Adding a new payment method now means writing one new function and adding one line to PAYMENT_STRATEGIES. The process_payment function never changes.
For the discount calculation example, the strategy pattern handles the tier-based cases cleanly:
LOYALTY_DISCOUNTS = {
"gold": 0.20,
"silver": 0.10,
"bronze": 0.05,
}
def calculate_discount(user, cart):
# Loyalty tier takes priority - O(1) lookup
if user.loyalty_tier in LOYALTY_DISCOUNTS:
return LOYALTY_DISCOUNTS[user.loyalty_tier]
# ...remaining rules handled by rule engine (Pattern 2)
The strategy pattern is a simplification of the Gang of Four Strategy pattern - the full OOP version involves abstract base classes and concrete strategy objects. For Python, a dictionary of functions is idiomatic and sufficient in most cases. Reserve class-based strategies for when strategies need to carry state or configuration.
Pattern 2: Rule Engine
The problem it solves: Multiple conditions where any number might match, they must be evaluated in a specific order, and new conditions must be addable without modifying the evaluation engine.
The mechanism: A list of (condition_function, action_function) tuples. The engine iterates the list, evaluates each condition, and calls the action of the first condition that matches (or all matching conditions, depending on the design).
# Discount rule engine - first matching rule wins
def make_rule(condition, discount):
"""Helper to create a (condition_fn, discount_value) rule."""
return (condition, discount)
def calculate_discount(user, cart):
rules = [
make_rule(lambda: user.loyalty_tier == "gold", 0.20),
make_rule(lambda: user.loyalty_tier == "silver", 0.10),
make_rule(lambda: user.loyalty_tier == "bronze", 0.05),
make_rule(lambda: user.is_first_purchase, 0.15),
make_rule(lambda: cart.total > 500, 0.08),
make_rule(lambda: cart.total > 200, 0.04),
make_rule(lambda: user.has_promo_code, cart.promo_discount),
make_rule(lambda: True, 0.0), # default
]
for condition, discount in rules:
if condition():
return discount
This version has one structural advantage over the if/elif version: the rules are data. They can be reordered by moving lines in the list. New rules can be inserted at any priority level. In a more sophisticated implementation, the rules themselves can come from a database or config file, allowing non-engineers to modify discount policy without a code deployment.
Making the Rule Engine Explicit and Testable
The inline lambda approach above couples the rules to the calling context. A more testable version separates rule definition from rule evaluation:
from dataclasses import dataclass
from typing import Callable, Any
@dataclass
class Rule:
name: str
condition: Callable[[], bool]
action: Callable[[], Any]
class RuleEngine:
def __init__(self, rules: list[Rule]):
self.rules = rules
def evaluate(self):
"""Evaluate rules in order; return the action result of the first match."""
for rule in self.rules:
if rule.condition():
return rule.action()
return None
def evaluate_all(self):
"""Evaluate all rules; return results from all matching actions."""
return [rule.action() for rule in self.rules if rule.condition()]
Usage:
def build_discount_rules(user, cart):
return [
Rule("gold_tier", lambda: user.loyalty_tier == "gold", lambda: 0.20),
Rule("silver_tier", lambda: user.loyalty_tier == "silver", lambda: 0.10),
Rule("first_buy", lambda: user.is_first_purchase, lambda: 0.15),
Rule("large_order", lambda: cart.total > 500, lambda: 0.08),
Rule("default", lambda: True, lambda: 0.0),
]
engine = RuleEngine(build_discount_rules(user, cart))
discount = engine.evaluate()
Each rule is now independently testable: construct a Rule object, call rule.condition() directly, verify the output. The engine's ordering logic is tested once, separately.
Production Example: Authorization as a Rule Engine
The authorization problem from the opening of this lesson is perfectly suited to the rule engine pattern:
# Authorization rule engine
# Rules are evaluated in priority order; first match determines access
def build_auth_rules(user, resource, action):
return [
Rule(
name="admin_allow_all",
condition=lambda: user.role == "admin",
action=lambda: True
),
Rule(
name="owner_allow_own",
condition=lambda: (user.role == "owner"
and resource.owner_id == user.id),
action=lambda: action in ("read", "write", "delete")
),
Rule(
name="member_read_public",
condition=lambda: (user.role == "member"
and resource.is_public),
action=lambda: action == "read"
),
Rule(
name="deny_all",
condition=lambda: True, # always matches - default
action=lambda: False
),
]
def can_user_act(user, resource, action):
engine = RuleEngine(build_auth_rules(user, resource, action))
return engine.evaluate()
Adding a new role or permission now means adding one Rule object to build_auth_rules. The engine itself is unchanged. The rules can be logged, audited, and tested independently.
Pattern 3: State Machine
The problem it solves: Objects that can be in one of several states, where transitions between states depend on events, and different behavior applies in different states.
The mechanism: States are represented as enums or strings. Transitions are defined as a dictionary mapping (current_state, event) → new_state. Actions (side effects) fire on transitions.
Without a state machine, state logic tends to produce nested conditionals scattered across the codebase:
# Ad-hoc - state checks everywhere
def process_event(order, event):
if order.status == "pending":
if event == "payment_received":
order.status = "paid"
send_confirmation(order)
elif event == "cancelled":
order.status = "cancelled"
send_cancellation(order)
else:
raise ValueError(f"Invalid event {event} for state {order.status}")
elif order.status == "paid":
if event == "shipped":
order.status = "shipped"
send_tracking_info(order)
elif event == "refund_requested":
order.status = "refund_pending"
else:
raise ValueError(f"Invalid event {event} for state {order.status}")
# ... more states ...
With an explicit state machine:
from enum import Enum
from typing import Callable
class OrderState(Enum):
PENDING = "pending"
PAID = "paid"
SHIPPED = "shipped"
REFUND = "refund_pending"
COMPLETE = "complete"
CANCELLED = "cancelled"
# Transition table: (current_state, event) → (new_state, action_fn)
def build_order_transitions():
return {
(OrderState.PENDING, "payment_received"): (OrderState.PAID, send_confirmation),
(OrderState.PENDING, "cancelled"): (OrderState.CANCELLED, send_cancellation),
(OrderState.PAID, "shipped"): (OrderState.SHIPPED, send_tracking_info),
(OrderState.PAID, "refund_requested"): (OrderState.REFUND, None),
(OrderState.SHIPPED, "delivered"): (OrderState.COMPLETE, send_delivery_receipt),
(OrderState.REFUND, "refund_approved"): (OrderState.CANCELLED, send_refund_confirmation),
}
class OrderStateMachine:
def __init__(self, order):
self.order = order
self.transitions = build_order_transitions()
def handle(self, event: str):
key = (self.order.state, event)
transition = self.transitions.get(key)
if transition is None:
raise ValueError(
f"No transition defined for state={self.order.state.value}, event={event}"
)
new_state, action_fn = transition
self.order.state = new_state
if action_fn is not None:
action_fn(self.order)
The state machine version has zero nested conditionals in the transition logic. All valid (state, event) pairs are listed in one place - the transition table. Invalid transitions raise an explicit error immediately. Adding a new state or event means adding one or two entries to the table, not modifying logic.
The transition table is also documentation - a complete visual spec of the state machine's behavior:
Pattern 4: Chain of Responsibility
The problem it solves: A request must be processed by one of several handlers, but which handler is appropriate depends on the request itself. Each handler decides whether it can handle the request; if not, it passes to the next.
The mechanism: An ordered sequence of handler objects or functions. Each handler either processes the request and stops the chain, or passes the request to the next handler.
# Logging system with fallback handlers
class ConsoleHandler:
def handle(self, log_level, message):
if log_level in ("DEBUG", "INFO"):
print(f"[{log_level}] {message}")
return True
return False # cannot handle - pass to next
class FileHandler:
def handle(self, log_level, message):
if log_level in ("WARNING", "ERROR"):
with open("app.log", "a") as f:
f.write(f"[{log_level}] {message}\n")
return True
return False
class AlertHandler:
def handle(self, log_level, message):
if log_level == "CRITICAL":
send_pagerduty_alert(message)
return True
return False
class LogChain:
def __init__(self, handlers):
self.handlers = handlers
def log(self, log_level, message):
for handler in self.handlers:
if handler.handle(log_level, message):
return
print(f"[UNHANDLED] {log_level}: {message}")
# Configuration
chain = LogChain([ConsoleHandler(), FileHandler(), AlertHandler()])
chain.log("ERROR", "Database connection lost") # → FileHandler
chain.log("CRITICAL", "System out of memory") # → AlertHandler
chain.log("DEBUG", "Cache hit for key user:42") # → ConsoleHandler
Alternatively, using functions in a list for a simpler version:
def handle_request(request, handlers):
"""Try each handler in order; stop at the first that returns a response."""
for handler in handlers:
response = handler(request)
if response is not None:
return response
return {"error": "No handler found"}, 404
Chain of Responsibility is the right pattern when: the set of handlers is dynamic (can be configured at runtime), each handler has independent logic for deciding whether to act, and you need to support different handler orderings for different contexts.
Pattern 5: Null Object
The problem it solves: Repeated if x is not None: checks scattered throughout code that consumes an optional object. The Null Object pattern replaces None with a do-nothing object that implements the same interface.
# Without Null Object: defensive checks everywhere
def render_dashboard(user):
if user.avatar is not None:
avatar_url = user.avatar.get_url()
else:
avatar_url = "/static/default_avatar.png"
if user.premium_features is not None:
features = user.premium_features.list()
else:
features = []
if user.notification_settings is not None:
notify = user.notification_settings.enabled
else:
notify = False
# With Null Object: zero None checks in the consumer
class NullAvatar:
def get_url(self):
return "/static/default_avatar.png"
class NullPremiumFeatures:
def list(self):
return []
class NullNotificationSettings:
@property
def enabled(self):
return False
def render_dashboard(user):
# avatar, premium_features, notification_settings are always set
# (to real objects or Null Objects) when the User is created
avatar_url = user.avatar.get_url() # works for real and NullAvatar
features = user.premium_features.list() # works for real and NullPremiumFeatures
notify = user.notification_settings.enabled
The Null Object is set when the User object is created:
class User:
def __init__(self, avatar=None, premium=None, notifications=None):
self.avatar = avatar or NullAvatar()
self.premium_features = premium or NullPremiumFeatures()
self.notification_settings = notifications or NullNotificationSettings()
Every call site that previously needed if x is not None: is now unconditional. The null-checking logic is centralized in one place (the User constructor) rather than scattered across every consumer.
Condition Functions as First-Class Objects
Python functions are objects. They can be stored in variables, passed as arguments, returned from functions, and composed. This capability is the foundation of all five patterns above, and it opens a powerful design space for condition systems.
# Predicates stored in variables
is_adult = lambda user: user.age >= 18
is_subscriber = lambda user: user.subscription_active
is_verified = lambda user: user.email_verified
# Predicates passed to higher-order functions
eligible_users = list(filter(is_adult, all_users))
verified_subscribers = list(filter(lambda u: is_subscriber(u) and is_verified(u), all_users))
# Predicate composition
import functools
def compose_and(*predicates):
"""Return a single predicate that is True only when ALL predicates are True."""
return functools.reduce(
lambda f, g: lambda x: f(x) and g(x),
predicates
)
def compose_or(*predicates):
"""Return a single predicate that is True when ANY predicate is True."""
return functools.reduce(
lambda f, g: lambda x: f(x) or g(x),
predicates
)
# Usage
can_access_content = compose_and(is_adult, is_subscriber, is_verified)
eligible_users = list(filter(can_access_content, all_users))
The compose_and function uses functools.reduce to fold a list of predicates into a single predicate. For two functions f and g, lambda x: f(x) and g(x) is a new function that returns True only if both return True. Reducing over a list of predicates chains this composition across all of them.
This pattern is particularly powerful for building configurable validation pipelines - see the Level 3 challenge.
Data-Driven Conditions
The final evolution: conditions that do not exist in code at all, but are parsed from configuration files or databases at runtime. This allows business rules to change without a code deployment.
import json
import operator
# Config file: business_rules.json
# {
# "discount_rules": [
# {"field": "loyalty_tier", "op": "eq", "value": "gold", "discount": 0.20},
# {"field": "cart_total", "op": "gt", "value": 500, "discount": 0.08}
# ]
# }
OPS = {
"eq": operator.eq,
"ne": operator.ne,
"gt": operator.gt,
"lt": operator.lt,
"ge": operator.ge,
"le": operator.le,
}
def load_rules(filepath):
with open(filepath) as f:
config = json.load(f)
return config["discount_rules"]
def evaluate_rule(rule, context):
"""Evaluate a single data-driven rule against a context dict."""
field_value = context.get(rule["field"])
comparator = OPS[rule["op"]]
return comparator(field_value, rule["value"])
def calculate_discount(user, cart, rules):
context = {
"loyalty_tier": user.loyalty_tier,
"cart_total": cart.total,
"is_first_purchase": user.is_first_purchase,
}
for rule in rules:
if evaluate_rule(rule, context):
return rule["discount"]
return 0.0
With this design, a product manager can change discount thresholds or add a new tier by editing a JSON file and reloading the application - no Python change, no code review, no deployment.
Data-driven conditions are powerful but carry risks. Untested rule configurations can cause incorrect behavior at runtime. Always validate loaded rules against a schema before using them, and run automated tests against representative rule sets as part of your CI pipeline.
Testing Each Pattern Independently
One major advantage of pattern-driven design is testability. Each component can be tested in isolation:
# Testing the rule engine - no need to test the full application stack
def test_gold_tier_gets_twenty_percent():
user = FakeUser(loyalty_tier="gold", is_first_purchase=False)
cart = FakeCart(total=100)
rules = build_discount_rules(user, cart)
engine = RuleEngine(rules)
assert engine.evaluate() == 0.20
def test_first_purchase_beats_large_order():
# First purchase rule comes before large-order rule
user = FakeUser(loyalty_tier=None, is_first_purchase=True)
cart = FakeCart(total=600)
rules = build_discount_rules(user, cart)
engine = RuleEngine(rules)
assert engine.evaluate() == 0.15 # first purchase, not large order
def test_state_machine_rejects_invalid_transition():
order = FakeOrder(state=OrderState.PENDING)
sm = OrderStateMachine(order)
with pytest.raises(ValueError):
sm.handle("shipped") # cannot ship a pending order
Contrast this with testing the ad-hoc nested version: every test must exercise the full function, and understanding why a test fails requires tracing through the entire nesting structure.
When NOT to Use Patterns
Every pattern adds abstraction. Abstraction has a cost: the code is harder to follow by tracing a single execution path, because the logic is distributed across multiple levels of indirection. Apply patterns only when the problem genuinely warrants the structure they provide.
Do not use the strategy pattern when you have two cases and no expectation of a third. A simple if/else is clearer, takes one line, and needs no documentation.
Do not use a rule engine when the rules are static, few in number, and unlikely to change. The overhead of constructing Rule objects and an engine is unjustified for three conditions.
Do not use a state machine when the object has two states and the transitions are trivial. A boolean flag is clearer.
Do not use the Null Object pattern when None is meaningful and distinguishable from the do-nothing case. The pattern breaks down when consumers need to know whether they are talking to a real object or a placeholder.
The general principle: choose the simplest design that accommodates current requirements and the next two likely changes. Over-engineering for hypothetical futures creates complexity that impedes the actual future.
The worst outcome of pattern-driven design is using a rule engine, a state machine, and the chain of responsibility pattern on a function that processes five static cases that will never change. Future developers will spend days understanding the infrastructure before discovering the logic is trivial. Match the pattern to the problem's actual complexity.
Interview Questions and Answers
Q1: What is the Strategy pattern in Python, and how does it differ from a simple if/elif chain?
The Strategy pattern replaces conditional branching driven by a discriminating value with a dictionary that maps that value to a callable function (the "strategy"). The if/elif version hardcodes every case into the function body, requiring modification of the function to add new cases and making cyclomatic complexity grow linearly with the number of cases. The strategy version has a constant cyclomatic complexity of 2 (one check for missing key), and adding a new case is a single-line addition to the dictionary - the evaluation function never changes. The dictionary is also inspectable and testable independently of the execution engine.
Q2: How does the rule engine pattern work, and what problem does it solve that a simple dispatch dictionary cannot?
A rule engine is an ordered list of (condition_function, action_function) pairs. The engine evaluates conditions in order and executes the first action whose condition returns True. The dispatch dictionary pattern requires that branching be driven by equality on a single value - it cannot handle complex conditions (range checks, multi-field conditions, computed predicates). The rule engine handles arbitrary conditions expressed as callable functions. It also makes rule priority explicit (the list order is the priority order) and allows the rule set to be extended, reordered, or loaded from external configuration without changing the engine code.
Q3: How do you implement a state machine in Python, and what are the advantages over nested conditionals?
A state machine is implemented with an enum (or string constants) for states, a dictionary mapping (current_state, event) tuples to (new_state, action_fn) pairs, and an engine that looks up and applies transitions. Advantages over nested conditionals: all valid transitions are listed in one place (the transition table), making the state space inspectable and documentable; invalid transitions are caught immediately at the lookup step rather than silently falling through; adding a new state or transition is a single entry in the table; and the engine logic (which is the same for all state machines) is tested once and never changes.
Q4: What is condition composition, and how do you use functools.reduce to combine predicates?
Condition composition is combining multiple predicate functions into a single predicate using logical operators. functools.reduce takes a binary function and a list, and applies the binary function cumulatively. For predicates, the binary function is lambda f, g: lambda x: f(x) and g(x) - a function that produces a new function that is True only when both f and g are True. Reducing a list of predicates with this combiner produces a single predicate equivalent to pred1(x) and pred2(x) and ... and predN(x). This allows predicates to be stored in lists, constructed dynamically, and combined without writing explicit conditional code.
Q5: When should you use each of the five patterns covered in this lesson?
Strategy: Value-dispatch branching (equality check on one variable), three or more cases, new cases expected. Rule Engine: Arbitrary condition functions evaluated in priority order, rules must be extensible or configurable, first-match or all-match semantics needed. State Machine: Object transitions between named states in response to events, invalid transitions must be detected, state logic is scattered across the codebase. Chain of Responsibility: A request must be handled by one of several handlers, each handler decides its own eligibility, the handler set is dynamic or ordered. Null Object: Optional objects checked for None repeatedly across many call sites, the do-nothing behavior is well-defined and uniform.
Q6: What is data-driven condition design, and what are its risks?
Data-driven conditions are conditions parsed from external sources (JSON config, database, feature flag service) at runtime rather than hardcoded in Python. The advantage is that business rules can change without code changes or deployments - non-engineers can modify rule configurations. The risks are: runtime errors from invalid rule configurations (mitigated by schema validation on load); inability to catch rule logic errors with static analysis; harder debugging when a rule produces unexpected results (no line numbers, no Python stack trace); and security risk if rule configurations can be modified by untrusted parties. Always validate loaded rules, log rule evaluation for debugging, and restrict who can modify rule configurations.
Graded Practice Challenges
Level 1 - Predict the Output
Given this rule engine, trace the output for the call evaluate_user(user) where user.age = 25, user.is_vip = False, user.has_coupon = True:
rules = [
(lambda u: u.is_vip, lambda u: f"VIP discount: 30%"),
(lambda u: u.age < 18, lambda u: f"Youth discount: 20%"),
(lambda u: u.has_coupon, lambda u: f"Coupon discount: 15%"),
(lambda u: u.age >= 65, lambda u: f"Senior discount: 25%"),
(lambda u: True, lambda u: f"No discount"),
]
def evaluate_user(user):
for condition, action in rules:
if condition(user):
return action(user)
Show Answer
Trace:
- Rule 1:
user.is_vip→False→ skip - Rule 2:
user.age < 18→25 < 18→False→ skip - Rule 3:
user.has_coupon→True→ match
Output: "Coupon discount: 15%"
Rules 4 and 5 are never evaluated because Rule 3 matched first. The senior discount rule (Rule 4) is unreachable for this user even though user.age = 25 does not satisfy it - the engine stops at the first match regardless. This illustrates why rule order matters critically in a first-match rule engine.
Level 2 - Debug the State Machine
This state machine for a traffic light has a bug. Identify it and provide the fix:
from enum import Enum
class Light(Enum):
RED = "red"
YELLOW = "yellow"
GREEN = "green"
TRANSITIONS = {
(Light.RED, "next"): Light.GREEN,
(Light.GREEN, "next"): Light.YELLOW,
(Light.YELLOW, "next"): Light.RED,
}
class TrafficLight:
def __init__(self):
self.state = Light.RED
def advance(self):
key = (self.state, "next")
new_state = TRANSITIONS[key] # KeyError if key missing
self.state = new_state
return self.state
# Scenario: what happens here?
light = TrafficLight()
light.state = Light.RED
light.advance() # RED → GREEN
light.advance() # GREEN → YELLOW
light.advance() # YELLOW → RED
light.advance() # RED → GREEN (should work)
print(light.state)
Show Answer
The scenario above actually works correctly - the transitions are a valid cycle.
The bug is using TRANSITIONS[key] (direct subscript) instead of TRANSITIONS.get(key). If any code sets light.state to a value not in the transition table, or if a new Light enum value is added without updating TRANSITIONS, the advance() call raises an unhandled KeyError with no diagnostic information.
Fix with proper error handling:
class TrafficLight:
def __init__(self):
self.state = Light.RED
def advance(self):
key = (self.state, "next")
new_state = TRANSITIONS.get(key)
if new_state is None:
raise ValueError(
f"No transition defined from state {self.state.value} on event 'next'"
)
self.state = new_state
return self.state
Additionally, the scenario should be tested with an invalid state to verify the error is raised and informative:
light = TrafficLight()
light.state = "purple" # invalid - not an enum member
try:
light.advance()
except (ValueError, KeyError) as e:
print(f"Error: {e}")
With the fix, ValueError is raised with a clear message. With the original code, KeyError is raised with a cryptic tuple key.
Level 3 - Design Challenge
Implement a configurable validation pipeline for user registration data. The pipeline must:
- Accept a list of validator functions at construction time - each validator takes a
dictand returns(bool, str)whereboolis whether the data is valid andstris an error message (empty string if valid) - Support three built-in validators:
validate_email,validate_password_strength,validate_age - Run all validators (not just until first failure) and collect all error messages
- Support adding validators at runtime via an
add_validator(fn)method - Include a
compose_validators(*fns)utility that combines multiple validators into one (the composite is valid only if all inner validators pass) - Write three test cases demonstrating: all validators pass, one validator fails, composite validator
Show Answer
import re
from typing import Callable
# Type alias for clarity
Validator = Callable[[dict], tuple[bool, str]]
# ── Built-in Validators ─────────────────────────────────────────────────
def validate_email(data: dict) -> tuple[bool, str]:
email = data.get("email", "")
pattern = r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
if not re.match(pattern, email):
return False, f"Invalid email address: '{email}'"
return True, ""
def validate_password_strength(data: dict) -> tuple[bool, str]:
password = data.get("password", "")
errors = []
if len(password) < 8:
errors.append("at least 8 characters")
if not any(c.isupper() for c in password):
errors.append("at least one uppercase letter")
if not any(c.isdigit() for c in password):
errors.append("at least one digit")
if errors:
return False, f"Password requires: {', '.join(errors)}"
return True, ""
def validate_age(data: dict) -> tuple[bool, str]:
age = data.get("age")
if age is None:
return False, "Age is required"
if not isinstance(age, int) or age < 0 or age > 150:
return False, f"Age must be an integer between 0 and 150, got: {age}"
if age < 13:
return False, f"Must be at least 13 years old, got: {age}"
return True, ""
# ── Validator Composition Utility ───────────────────────────────────────
def compose_validators(*validators: Validator) -> Validator:
"""
Combine multiple validators into a single validator.
The composite is valid only if ALL inner validators pass.
All error messages from failing validators are collected.
"""
def composite(data: dict) -> tuple[bool, str]:
all_errors = []
for validator in validators:
is_valid, error = validator(data)
if not is_valid:
all_errors.append(error)
if all_errors:
return False, "; ".join(all_errors)
return True, ""
return composite
# ── Validation Pipeline ─────────────────────────────────────────────────
class ValidationPipeline:
"""
Runs all registered validators against input data.
Collects ALL errors (does not stop at first failure).
Supports dynamic registration of validators.
"""
def __init__(self, validators: list[Validator] = None):
self._validators: list[Validator] = list(validators or [])
def add_validator(self, validator: Validator) -> "ValidationPipeline":
"""Add a validator at runtime. Returns self for chaining."""
self._validators.append(validator)
return self
def validate(self, data: dict) -> tuple[bool, list[str]]:
"""
Run all validators against data.
Returns (all_valid, list_of_error_messages).
"""
errors = []
for validator in self._validators:
is_valid, error = validator(data)
if not is_valid:
errors.append(error)
return len(errors) == 0, errors
# ── Test Cases ──────────────────────────────────────────────────────────
def test_all_validators_pass():
pipeline = ValidationPipeline([
validate_email,
validate_password_strength,
validate_age,
])
is_valid, errors = pipeline.validate(data)
assert is_valid is True
assert errors == []
print("PASS: test_all_validators_pass")
def test_one_validator_fails():
pipeline = ValidationPipeline([
validate_email,
validate_password_strength,
validate_age,
])
is_valid, errors = pipeline.validate(data)
assert is_valid is False
assert len(errors) == 1
assert "8 characters" in errors[0]
print("PASS: test_one_validator_fails")
def test_composite_validator():
# Composite: email AND password must both pass for the composite to pass
email_and_password = compose_validators(validate_email, validate_password_strength)
pipeline = ValidationPipeline([email_and_password, validate_age])
data = {"email": "bad-email", "password": "weak", "age": 25}
is_valid, errors = pipeline.validate(data)
assert is_valid is False
# Composite produces ONE error message containing both failures
assert len(errors) == 1
assert "Invalid email" in errors[0]
assert "8 characters" in errors[0]
print("PASS: test_composite_validator")
def test_dynamic_validator_addition():
pipeline = ValidationPipeline([validate_email])
# Adding validator at runtime
pipeline.add_validator(validate_age)
is_valid, errors = pipeline.validate(data)
assert is_valid is False
assert any("13 years" in e for e in errors)
print("PASS: test_dynamic_validator_addition")
# Run all tests
test_all_validators_pass()
test_one_validator_fails()
test_composite_validator()
test_dynamic_validator_addition()
Design decisions explained:
Validator = Callable[[dict], tuple[bool, str]]- the type alias documents the contract every validator must fulfill, enabling type checkers to catch violations.compose_validatorsuses the same composition principle asfunctools.reduceover predicates, but collects all errors rather than short-circuiting - appropriate for validation (you want to show all errors at once, not one at a time).ValidationPipelineruns all validators rather than stopping at the first failure - this is the "collect all errors" mode, appropriate for form validation where users need to see and fix all problems at once.add_validatorreturnsselfto support chaining:pipeline.add_validator(v1).add_validator(v2).- Each validator is independently testable - no pipeline needed, just call
validate_email(data)directly.
Quick Reference Cheatsheet
| Pattern | Structure | Best For | When NOT to Use |
|---|---|---|---|
| Strategy | {key: fn} dict | Value-dispatch, 3+ cases | 2 cases, no growth expected |
| Rule Engine | [(cond_fn, action_fn)] list | Arbitrary conditions, ordered priority | Few static rules |
| State Machine | {(state, event): new_state} dict | Object lifecycle, transitions | 2 states, trivial transitions |
| Chain of Responsibility | [handler, handler, ...] list | Request routing, dynamic handlers | Simple linear branching |
| Null Object | Default class implementing interface | Optional objects checked everywhere | When None carries distinct meaning |
| Composition Tool | Purpose |
|---|---|
lambda x: f(x) and g(x) | Combine two predicates with AND |
functools.reduce(combiner, predicates) | Fold N predicates into one |
compose_validators(*fns) | Pipeline-style validator composition |
filter(predicate, iterable) | Apply predicate to collection |
all(pred(x) for x in items) | Check all items satisfy predicate |
Key Takeaways
- Ad-hoc
if/elifchains embed rules in logic; pattern-driven design separates rules from the evaluation machinery, making rules extensible without modifying the engine. - The Strategy pattern maps discriminating values to callable functions - adding a new case is one line in the dictionary, not a structural change to the function.
- The Rule Engine pattern stores conditions as callable functions in an ordered list - conditions can be arbitrary (not just equality checks), and rule priority is explicit via list order.
- The State Machine pattern makes state transitions explicit and exhaustive in a single transition table, catching invalid transitions immediately and documenting the full state space in one place.
- Chain of Responsibility delegates request handling to an ordered sequence of handlers, each of which decides its own eligibility - appropriate when the handler set is dynamic or context-dependent.
- The Null Object pattern centralizes None-handling at object creation time, eliminating scattered defensive
if x is not None:checks throughout consumer code. - Condition functions are first-class objects in Python - they can be stored in variables, passed to higher-order functions, and composed with
functools.reduceto build composite predicates dynamically. - Data-driven conditions (rules from config files or databases) enable rule changes without code deployments, but require schema validation, comprehensive testing of rule configurations, and careful access control.
- Every pattern adds abstraction with a cost: choose the simplest design that accommodates current requirements and realistic future changes - over-engineering static, simple logic is as harmful as under-engineering dynamic, complex logic.
