Python Pattern-Driven Condition Design: Practice Problems & Exercises
Practice: Pattern-Driven Condition Design
← Back to lessonEasy
Refactor an if/elif/else chain that maps HTTP status codes to messages into a clean dictionary dispatch table. Handle unknown codes with a default.
def get_status_message(code):
"""Return HTTP status message using dict dispatch."""
status_map = {
200: "OK",
301: "Moved Permanently",
404: "Not Found",
500: "Internal Server Error",
}
return status_map.get(code, "Unknown Status")
for code in [200, 301, 404, 500, 418]:
print(f"{code}: {get_status_message(code)}")Solution
def get_status_message(code):
"""Return HTTP status message using dict dispatch."""
status_map = {
200: "OK",
301: "Moved Permanently",
404: "Not Found",
500: "Internal Server Error",
}
return status_map.get(code, "Unknown Status")
for code in [200, 301, 404, 500, 418]:
print(f"{code}: {get_status_message(code)}")
Output:
200: OK
301: Moved Permanently
404: Not Found
500: Internal Server Error
418: Unknown Status
Why this is better: The dictionary approach is O(1) lookup vs O(n) sequential comparison in if/elif chains. Adding a new status code is a single line in the dictionary rather than a new elif branch. The dict.get(key, default) method handles the fallback cleanly without needing an explicit else block.
Key insight: Whenever you see an if/elif chain that maps one value to another (input to output) with no complex logic in each branch, a dictionary dispatch is almost always the cleaner pattern. It separates data from control flow.
def get_status_message_old(code):
"""Old approach — if/elif chain."""
if code == 200:
return "OK"
elif code == 301:
return "Moved Permanently"
elif code == 404:
return "Not Found"
elif code == 500:
return "Internal Server Error"
else:
return "Unknown Status"
def get_status_message(code):
"""Refactor using a dictionary dispatch table."""
# TODO: Replace the if/elif chain with a dict lookup
pass
# Test
for code in [200, 301, 404, 500, 418]:
print(f"{code}: {get_status_message(code)}")Expected Output
200: OK\n301: Moved Permanently\n404: Not Found\n500: Internal Server Error\n418: Unknown StatusHints
Hint 1: Create a dictionary mapping status codes to their messages.
Hint 2: Use dict.get(key, default) to handle unknown codes with a fallback value.
Implement a strategy selector that picks a sorting function by name. Store the strategy functions in a dictionary and look up the right one at runtime.
def sort_by_name(items):
return sorted(items)
def sort_by_length(items):
return sorted(items, key=len)
def sort_by_last_char(items):
return sorted(items, key=lambda x: x[-1])
def sort_items(items, strategy="name"):
"""Select a sorting strategy by name and apply it."""
strategies = {
"name": sort_by_name,
"length": sort_by_length,
"last_char": sort_by_last_char,
}
fn = strategies.get(strategy, sort_by_name)
return fn(items)
fruits = ["banana", "kiwi", "apple", "dragonfruit", "fig"]
for strategy in ["name", "length", "last_char", "unknown"]:
result = sort_items(fruits, strategy)
print(f"{strategy:>10}: {result}")Solution
def sort_by_name(items):
return sorted(items)
def sort_by_length(items):
return sorted(items, key=len)
def sort_by_last_char(items):
return sorted(items, key=lambda x: x[-1])
def sort_items(items, strategy="name"):
"""Select a sorting strategy by name and apply it."""
strategies = {
"name": sort_by_name,
"length": sort_by_length,
"last_char": sort_by_last_char,
}
fn = strategies.get(strategy, sort_by_name)
return fn(items)
fruits = ["banana", "kiwi", "apple", "dragonfruit", "fig"]
for strategy in ["name", "length", "last_char", "unknown"]:
result = sort_items(fruits, strategy)
print(f"{strategy:>10}: {result}")
Output:
name: ['apple', 'banana', 'dragonfruit', 'fig', 'kiwi']
length: ['fig', 'kiwi', 'apple', 'banana', 'dragonfruit']
last_char: ['banana', 'apple', 'dragonfruit', 'fig', 'kiwi']
unknown: ['apple', 'banana', 'dragonfruit', 'fig', 'kiwi']
How it works: This is the Strategy Pattern — one of the most useful design patterns in Python. Instead of branching with if/elif to pick behavior, you store callable strategies in a dictionary and select at runtime. The strategies.get(strategy, sort_by_name) call returns the function object (not its result), and then fn(items) invokes it.
Key insight: The Strategy Pattern decouples what to do from how to choose. Adding a new strategy means writing a new function and adding one dictionary entry — zero changes to the selection logic.
def sort_by_name(items):
return sorted(items)
def sort_by_length(items):
return sorted(items, key=len)
def sort_by_last_char(items):
return sorted(items, key=lambda x: x[-1])
def sort_items(items, strategy="name"):
"""Select a sorting strategy by name and apply it.
Strategies: 'name', 'length', 'last_char'
Default to sort_by_name for unknown strategies.
"""
# TODO: Use a dict to map strategy names to functions
pass
fruits = ["banana", "kiwi", "apple", "dragonfruit", "fig"]
for strategy in ["name", "length", "last_char", "unknown"]:
result = sort_items(fruits, strategy)
print(f"{strategy:>10}: {result}")Expected Output
name: ['apple', 'banana', 'dragonfruit', 'fig', 'kiwi']\n length: ['fig', 'kiwi', 'apple', 'banana', 'dragonfruit']\n last_char: ['banana', 'apple', 'dragonfruit', 'fig', 'kiwi']\n unknown: ['apple', 'banana', 'dragonfruit', 'fig', 'kiwi']Hints
Hint 1: Functions are first-class objects in Python — you can store them in a dictionary as values.
Hint 2: Map strategy name strings to the actual function objects (without calling them).
Build a calculator that dispatches math operations using a dictionary of lambda functions. Handle division by zero and unknown operations gracefully.
def calculate(a, b, operation):
"""Perform a math operation using callable dispatch."""
ops = {
"add": lambda x, y: x + y,
"sub": lambda x, y: x - y,
"mul": lambda x, y: x * y,
"div": lambda x, y: x / y,
"mod": lambda x, y: x % y,
"pow": lambda x, y: x ** y,
}
fn = ops.get(operation)
if fn is None:
return "Error: unknown operation"
try:
return fn(a, b)
except ZeroDivisionError:
return "Error: division by zero"
test_cases = [
(10, 3, "add"),
(10, 3, "sub"),
(10, 3, "mul"),
(10, 3, "div"),
(10, 3, "mod"),
(2, 10, "pow"),
(10, 0, "div"),
(10, 3, "sqrt"),
]
for a, b, op in test_cases:
print(f"{a} {op} {b} = {calculate(a, b, op)}")Solution
def calculate(a, b, operation):
"""Perform a math operation using callable dispatch."""
ops = {
"add": lambda x, y: x + y,
"sub": lambda x, y: x - y,
"mul": lambda x, y: x * y,
"div": lambda x, y: x / y,
"mod": lambda x, y: x % y,
"pow": lambda x, y: x ** y,
}
fn = ops.get(operation)
if fn is None:
return "Error: unknown operation"
try:
return fn(a, b)
except ZeroDivisionError:
return "Error: division by zero"
test_cases = [
(10, 3, "add"),
(10, 3, "sub"),
(10, 3, "mul"),
(10, 3, "div"),
(10, 3, "mod"),
(2, 10, "pow"),
(10, 0, "div"),
(10, 3, "sqrt"),
]
for a, b, op in test_cases:
print(f"{a} {op} {b} = {calculate(a, b, op)}")
Output:
10 add 3 = 13
10 sub 3 = 7
10 mul 3 = 30
10 div 3 = 3.3333333333333335
10 mod 3 = 1
2 pow 10 = 1024
10 div 0 = Error: division by zero
10 sqrt 3 = Error: unknown operation
How it works: Each operation is a lambda stored in the ops dictionary. The lookup is O(1), and adding new operations requires zero changes to control flow — just add a new key-value pair. Error handling is centralized in one try/except block rather than being scattered across elif branches.
Key insight: By separating the dispatch lookup (ops.get) from the error handling (try/except), the code has clear, distinct responsibilities. The None check for unknown operations happens before any computation, following the fail-fast principle.
def calculate(a, b, operation):
"""Perform a math operation using callable dispatch.
Operations: 'add', 'sub', 'mul', 'div', 'mod', 'pow'
Return 'Error: unknown operation' for invalid operations.
Return 'Error: division by zero' if dividing/mod by zero.
"""
# TODO: Build a dispatch table using lambdas
pass
# Test
test_cases = [
(10, 3, "add"),
(10, 3, "sub"),
(10, 3, "mul"),
(10, 3, "div"),
(10, 3, "mod"),
(2, 10, "pow"),
(10, 0, "div"),
(10, 3, "sqrt"),
]
for a, b, op in test_cases:
print(f"{a} {op} {b} = {calculate(a, b, op)}")Expected Output
10 add 3 = 13\n10 sub 3 = 7\n10 mul 3 = 30\n10 div 3 = 3.3333333333333335\n10 mod 3 = 1\n2 pow 10 = 1024\n10 div 0 = Error: division by zero\n10 sqrt 3 = Error: unknown operationHints
Hint 1: Use lambdas for simple one-liner operations: lambda a, b: a + b
Hint 2: Wrap the function call in a try/except to catch ZeroDivisionError.
Medium
Build a file format handler that dispatches to the correct parser based on file extension. Support CSV, JSON, and INI formats, with graceful handling of unknown types.
import json
import csv
import io
def parse_csv_data(content):
"""Parse CSV string into list of dicts."""
reader = csv.DictReader(io.StringIO(content))
return list(reader)
def parse_json_data(content):
"""Parse JSON string into Python object."""
return json.loads(content)
def parse_ini_data(content):
"""Parse simple INI-style key=value content into dict."""
result = {}
for line in content.strip().split("\n"):
if "=" in line:
key, value = line.split("=", 1)
result[key.strip()] = value.strip()
return result
def process_file(filename, content):
"""Dispatch to the correct parser based on file extension."""
parsers = {
"csv": parse_csv_data,
"json": parse_json_data,
"ini": parse_ini_data,
}
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
parser = parsers.get(ext)
if parser is None:
return {"format": "unknown", "data": None, "records": 0}
data = parser(content)
records = len(data) if isinstance(data, (list, dict)) else 0
return {"format": ext, "data": data, "records": records}
# Test data
csv_content = "name,age,role\nAlice,30,Engineer\nBob,25,Designer\nCharlie,35,Manager"
json_content = json.dumps([{"id": 1, "status": "active"}, {"id": 2, "status": "inactive"}])
ini_content = "host=localhost\nport=5432\ndb=myapp\nuser=admin"
test_files = [
("employees.csv", csv_content),
("config.json", json_content),
("settings.ini", ini_content),
("readme.txt", "Hello world"),
]
for filename, content in test_files:
result = process_file(filename, content)
print(f"{filename:>16} -> format={result['format']}, records={result['records']}, type={type(result['data']).__name__}")Solution
import json
import csv
import io
def parse_csv_data(content):
reader = csv.DictReader(io.StringIO(content))
return list(reader)
def parse_json_data(content):
return json.loads(content)
def parse_ini_data(content):
result = {}
for line in content.strip().split("\n"):
if "=" in line:
key, value = line.split("=", 1)
result[key.strip()] = value.strip()
return result
def process_file(filename, content):
parsers = {
"csv": parse_csv_data,
"json": parse_json_data,
"ini": parse_ini_data,
}
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
parser = parsers.get(ext)
if parser is None:
return {"format": "unknown", "data": None, "records": 0}
data = parser(content)
records = len(data) if isinstance(data, (list, dict)) else 0
return {"format": ext, "data": data, "records": records}
Output:
employees.csv -> format=csv, records=3, type=list
config.json -> format=json, records=2, type=list
settings.ini -> format=ini, records=4, type=dict
readme.txt -> format=unknown, records=0, type=NoneType
Why dispatch beats if/elif here: Each parser has completely independent logic — there is no shared state or sequential dependency between them. The dispatch table makes it trivial to add a new format (write a parser function, add one dict entry). With if/elif, you would need to find the right place in the chain and add another branch, risking accidental fall-through logic.
Key insight: The process_file function never needs to know how each format is parsed. It only knows which parser to call. This separation of concerns makes each parser independently testable and the dispatcher trivially extensible.
import json
import csv
import io
def parse_csv_data(content):
"""Parse CSV string into list of dicts."""
# TODO: implement
pass
def parse_json_data(content):
"""Parse JSON string into Python object."""
# TODO: implement
pass
def parse_ini_data(content):
"""Parse simple INI-style key=value content into dict."""
# TODO: implement
pass
def process_file(filename, content):
"""Dispatch to the correct parser based on file extension.
Return dict with 'format', 'data', and 'records' (count of items).
For unknown formats, return format='unknown', data=None, records=0.
"""
# TODO: Extract extension, dispatch to correct parser
pass
# Test data
csv_content = "name,age,role\nAlice,30,Engineer\nBob,25,Designer\nCharlie,35,Manager"
json_content = json.dumps([{"id": 1, "status": "active"}, {"id": 2, "status": "inactive"}])
ini_content = "host=localhost\nport=5432\ndb=myapp\nuser=admin"
test_files = [
("employees.csv", csv_content),
("config.json", json_content),
("settings.ini", ini_content),
("readme.txt", "Hello world"),
]
for filename, content in test_files:
result = process_file(filename, content)
print(f"{filename:>16} -> format={result['format']}, records={result['records']}, type={type(result['data']).__name__}")Expected Output
employees.csv -> format=csv, records=3, type=list\n config.json -> format=json, records=2, type=list\n settings.ini -> format=ini, records=4, type=dict\n readme.txt -> format=unknown, records=0, type=NoneTypeHints
Hint 1: Extract the file extension with filename.rsplit(".", 1) to get the part after the last dot.
Hint 2: csv.DictReader can parse CSV strings if you wrap them in io.StringIO.
Hint 3: For INI parsing, split each line on "=" — no need for configparser for simple key=value.
Build a notification router that dispatches messages to email, SMS, or push handlers based on channel name. Each handler formats the message differently and accepts channel-specific keyword arguments.
def send_email(recipient, message, **kwargs):
subject = kwargs.get("subject", "Notification")
return f"[EMAIL] To: {recipient} | Subject: {subject} | Body: {message}"
def send_sms(recipient, message, **kwargs):
truncated = message[:160]
return f"[SMS] To: {recipient} | Body (160 chars max): {truncated}"
def send_push(recipient, message, **kwargs):
priority = kwargs.get("priority", "normal")
return f"[PUSH] To: {recipient} | Priority: {priority} | Body: {message}"
def notify(channel, recipient, message, **kwargs):
"""Route notification to the correct channel."""
routers = {
"email": send_email,
"sms": send_sms,
"push": send_push,
}
handler = routers.get(channel)
if handler is None:
return f"Error: unknown channel '{channel}'"
return handler(recipient, message, **kwargs)
notifications = [
("email", "[email protected]", "Your order shipped!", {"subject": "Order Update"}),
("sms", "+1234567890", "Code: 847293", {}),
("push", "user_42", "New message from Bob", {"priority": "high"}),
("email", "[email protected]", "Meeting at 3pm", {}),
("slack", "#general", "Deploy complete", {}),
]
for channel, recipient, message, kwargs in notifications:
result = notify(channel, recipient, message, **kwargs)
print(result)Solution
def send_email(recipient, message, **kwargs):
subject = kwargs.get("subject", "Notification")
return f"[EMAIL] To: {recipient} | Subject: {subject} | Body: {message}"
def send_sms(recipient, message, **kwargs):
truncated = message[:160]
return f"[SMS] To: {recipient} | Body (160 chars max): {truncated}"
def send_push(recipient, message, **kwargs):
priority = kwargs.get("priority", "normal")
return f"[PUSH] To: {recipient} | Priority: {priority} | Body: {message}"
def notify(channel, recipient, message, **kwargs):
routers = {
"email": send_email,
"sms": send_sms,
"push": send_push,
}
handler = routers.get(channel)
if handler is None:
return f"Error: unknown channel '{channel}'"
return handler(recipient, message, **kwargs)
Output:
[EMAIL] To: [email protected] | Subject: Order Update | Body: Your order shipped!
[SMS] To: +1234567890 | Body (160 chars max): Code: 847293
[PUSH] To: user_42 | Priority: high | Body: New message from Bob
[EMAIL] To: [email protected] | Subject: Notification | Body: Meeting at 3pm
Error: unknown channel 'slack'
Why **kwargs matters here: Each notification channel has different parameters (email needs a subject, push needs a priority, SMS has none). Using **kwargs lets the dispatcher pass all options through without knowing what each handler expects. This is the Open/Closed Principle in action — the notify function is open for extension (new channels) but closed for modification (existing code never changes).
Key insight: The combination of dispatch tables with **kwargs is extremely powerful in Python. The dispatcher provides uniform routing while each handler can accept whatever parameters it needs. This pattern appears everywhere in production code — web frameworks, logging systems, message queues.
def send_email(recipient, message, **kwargs):
subject = kwargs.get("subject", "Notification")
# TODO: Return formatted string showing email delivery
pass
def send_sms(recipient, message, **kwargs):
truncated = message[:160]
# TODO: Return formatted string showing SMS delivery
pass
def send_push(recipient, message, **kwargs):
priority = kwargs.get("priority", "normal")
# TODO: Return formatted string showing push delivery
pass
def notify(channel, recipient, message, **kwargs):
"""Route notification to the correct channel.
Supported: 'email', 'sms', 'push'
Return 'Error: unknown channel <channel>' for unsupported channels.
"""
# TODO: Dispatch to the correct sender
pass
# Test
notifications = [
("email", "[email protected]", "Your order shipped!", {"subject": "Order Update"}),
("sms", "+1234567890", "Code: 847293", {}),
("push", "user_42", "New message from Bob", {"priority": "high"}),
("email", "[email protected]", "Meeting at 3pm", {}),
("slack", "#general", "Deploy complete", {}),
]
for channel, recipient, message, kwargs in notifications:
result = notify(channel, recipient, message, **kwargs)
print(result)Expected Output
[EMAIL] To: [email protected] | Subject: Order Update | Body: Your order shipped!\n[SMS] To: +1234567890 | Body (160 chars max): Code: 847293\n[PUSH] To: user_42 | Priority: high | Body: New message from Bob\n[EMAIL] To: [email protected] | Subject: Notification | Body: Meeting at 3pm\nError: unknown channel 'slack'Hints
Hint 1: Each sender function should return a formatted string (not print it) so the dispatcher can handle it uniformly.
Hint 2: Use **kwargs to pass channel-specific options (subject for email, priority for push) without the dispatcher knowing about them.
Build a validator registry that maps field names to lists of validation functions. Each validator returns None for valid or an error string for invalid. Support multiple validators per field and collect all errors.
class ValidatorRegistry:
def __init__(self):
self._validators = {}
def register(self, field_name, validator_fn):
if field_name not in self._validators:
self._validators[field_name] = []
self._validators[field_name].append(validator_fn)
def validate(self, data):
errors = {}
for field, validators in self._validators.items():
value = data.get(field)
for validator_fn in validators:
error = validator_fn(value)
if error is not None:
if field not in errors:
errors[field] = []
errors[field].append(error)
return errors
def required(value):
if not value and value != 0:
return "is required"
return None
def min_length(n):
def check(value):
if isinstance(value, str) and len(value) < n:
return f"must be at least {n} characters"
return None
return check
def is_positive(value):
if isinstance(value, (int, float)) and value <= 0:
return "must be positive"
return None
def is_email(value):
if isinstance(value, str) and "@" not in value:
return "must be a valid email"
return None
registry = ValidatorRegistry()
registry.register("name", required)
registry.register("name", min_length(2))
registry.register("email", required)
registry.register("email", is_email)
registry.register("age", required)
registry.register("age", is_positive)
test_records = [
{"name": "Alice", "email": "[email protected]", "age": 30},
{"name": "", "email": "not-an-email", "age": -5},
{"name": "A", "email": "", "age": 0},
]
for i, record in enumerate(test_records):
errors = registry.validate(record)
if errors:
print(f"Record {i}: INVALID")
for field, msgs in sorted(errors.items()):
for msg in msgs:
print(f" {field} {msg}")
else:
print(f"Record {i}: VALID")Solution
class ValidatorRegistry:
def __init__(self):
self._validators = {}
def register(self, field_name, validator_fn):
if field_name not in self._validators:
self._validators[field_name] = []
self._validators[field_name].append(validator_fn)
def validate(self, data):
errors = {}
for field, validators in self._validators.items():
value = data.get(field)
for validator_fn in validators:
error = validator_fn(value)
if error is not None:
if field not in errors:
errors[field] = []
errors[field].append(error)
return errors
Output:
Record 0: VALID
Record 1: INVALID
age must be positive
email must be a valid email
name is required
Record 2: INVALID
age is required
email is required
email must be a valid email
name must be at least 2 characters
Why the registry pattern: The validator registry decouples validation rules from the validation engine. The validate method has zero knowledge of what "required" or "is_email" means — it just calls functions and collects results. New validators can be added without modifying any existing code.
The closure trick: min_length(n) is a factory that returns a closure. The inner check function captures n from the enclosing scope. This lets you create parameterized validators (min_length(2), min_length(10)) that all share the same interface: take a value, return None or an error string.
Key insight: This is how real-world validation libraries work (Django forms, Pydantic, marshmallow). The registry pattern combined with factory functions gives you composable, reusable validation with zero if/elif chains.
class ValidatorRegistry:
"""A registry that maps field names to validation functions."""
def __init__(self):
self._validators = {}
def register(self, field_name, validator_fn):
"""Register a validator function for a field."""
# TODO: Store the validator. Support multiple validators per field.
pass
def validate(self, data):
"""Validate a data dict. Return dict of {field: [errors]}."""
# TODO: Run all validators for each field, collect errors.
pass
# Define validators
def required(value):
if not value and value != 0:
return "is required"
return None
def min_length(n):
def check(value):
if isinstance(value, str) and len(value) < n:
return f"must be at least {n} characters"
return None
return check
def is_positive(value):
if isinstance(value, (int, float)) and value <= 0:
return "must be positive"
return None
def is_email(value):
if isinstance(value, str) and "@" not in value:
return "must be a valid email"
return None
# Build registry
registry = ValidatorRegistry()
registry.register("name", required)
registry.register("name", min_length(2))
registry.register("email", required)
registry.register("email", is_email)
registry.register("age", required)
registry.register("age", is_positive)
# Test with valid and invalid data
test_records = [
{"name": "Alice", "email": "[email protected]", "age": 30},
{"name": "", "email": "not-an-email", "age": -5},
{"name": "A", "email": "", "age": 0},
]
for i, record in enumerate(test_records):
errors = registry.validate(record)
if errors:
print(f"Record {i}: INVALID")
for field, msgs in sorted(errors.items()):
for msg in msgs:
print(f" {field} {msg}")
else:
print(f"Record {i}: VALID")Expected Output
Record 0: VALID\nRecord 1: INVALID\n age must be positive\n email must be a valid email\n name is required\nRecord 2: INVALID\n age is required\n email is required\n email must be a valid email\n name must be at least 2 charactersHints
Hint 1: Use a dict where each key maps to a list of validator functions — one field can have multiple validators.
Hint 2: A validator returns None if the value is valid, or an error message string if invalid.
Hint 3: min_length(n) is a factory function — it returns a closure that captures n.
Implement a chain of responsibility where a request passes through authentication, rate limiting, and validation handlers in sequence. If any handler rejects the request, the chain stops immediately.
def auth_handler(request):
"""Check authentication. Return error dict or None to continue."""
if not request.get("auth_token"):
return {"status": 401, "message": "Authentication required"}
return None
def rate_limit_handler(request):
"""Check rate limit. Return error dict or None to continue."""
if request.get("requests_count", 0) >= 100:
return {"status": 429, "message": "Rate limit exceeded"}
return None
def validate_handler(request):
"""Validate request body. Return error dict or None to continue."""
body = request.get("body")
if not isinstance(body, dict) or not body:
return {"status": 400, "message": "Request body must be a non-empty dict"}
return None
def process_request(request):
"""Process a valid request. Return success dict."""
return {"status": 200, "message": f"Processed: {request['body']}"}
def handle_request(request):
"""Run request through a chain of handlers."""
chain = [auth_handler, rate_limit_handler, validate_handler]
for handler in chain:
result = handler(request)
if result is not None:
return result
return process_request(request)
requests_list = [
{"auth_token": "abc123", "requests_count": 50, "body": {"action": "create"}},
{"requests_count": 50, "body": {"action": "create"}},
{"auth_token": "abc123", "requests_count": 150, "body": {"action": "create"}},
{"auth_token": "abc123", "requests_count": 50, "body": {}},
{"auth_token": "abc123", "requests_count": 50},
]
for i, req in enumerate(requests_list):
result = handle_request(req)
print(f"Request {i}: [{result['status']}] {result['message']}")Solution
def auth_handler(request):
if not request.get("auth_token"):
return {"status": 401, "message": "Authentication required"}
return None
def rate_limit_handler(request):
if request.get("requests_count", 0) >= 100:
return {"status": 429, "message": "Rate limit exceeded"}
return None
def validate_handler(request):
body = request.get("body")
if not isinstance(body, dict) or not body:
return {"status": 400, "message": "Request body must be a non-empty dict"}
return None
def process_request(request):
return {"status": 200, "message": f"Processed: {request['body']}"}
def handle_request(request):
chain = [auth_handler, rate_limit_handler, validate_handler]
for handler in chain:
result = handler(request)
if result is not None:
return result
return process_request(request)
Output:
Request 0: [200] Processed: {'action': 'create'}
Request 1: [401] Authentication required
Request 2: [429] Rate limit exceeded
Request 3: [400] Request body must be a non-empty dict
Request 4: [400] Request body must be a non-empty dict
How the chain works: Each handler has a simple contract: return None to pass the request to the next handler, or return an error dict to short-circuit the chain. The handle_request function iterates through the chain list and stops at the first non-None response.
This is middleware in disguise: This exact pattern powers middleware stacks in Flask, Django, Express.js, and every major web framework. Each middleware layer checks one concern (auth, rate limiting, CORS, logging) and either passes the request through or returns an error.
Key insight: The chain of responsibility pattern transforms deeply nested if/elif logic into a flat, ordered list of independent checks. Each handler is self-contained, testable in isolation, and the chain order can be changed by reordering the list.
def auth_handler(request):
"""Check authentication. Return error dict or None to continue."""
# TODO: Check if 'auth_token' is in request and not empty
pass
def rate_limit_handler(request):
"""Check rate limit. Return error dict or None to continue."""
# TODO: Check if 'requests_count' < 100
pass
def validate_handler(request):
"""Validate request body. Return error dict or None to continue."""
# TODO: Check if 'body' is in request and is a non-empty dict
pass
def process_request(request):
"""Process a valid request. Return success dict."""
# TODO: Return success response with the body
pass
def handle_request(request):
"""Run request through a chain of handlers.
Each handler returns an error dict to stop the chain, or None to continue.
If all handlers pass, process the request.
"""
# TODO: Build the chain and process
pass
# Test
requests_list = [
{"auth_token": "abc123", "requests_count": 50, "body": {"action": "create"}},
{"requests_count": 50, "body": {"action": "create"}},
{"auth_token": "abc123", "requests_count": 150, "body": {"action": "create"}},
{"auth_token": "abc123", "requests_count": 50, "body": {}},
{"auth_token": "abc123", "requests_count": 50},
]
for i, req in enumerate(requests_list):
result = handle_request(req)
print(f"Request {i}: [{result['status']}] {result['message']}")Expected Output
Request 0: [200] Processed: {'action': 'create'}\nRequest 1: [401] Authentication required\nRequest 2: [429] Rate limit exceeded\nRequest 3: [400] Request body must be a non-empty dict\nRequest 4: [400] Request body must be a non-empty dictHints
Hint 1: Each handler in the chain should return a dict with "status" and "message" keys on failure, or None on success.
Hint 2: Store the handlers in a list and iterate — if any handler returns a non-None result, stop and return it immediately.
Hard
Build a command processor where functions self-register as commands using a decorator. The processor parses command strings, dispatches to the right handler, and provides built-in help.
class CommandProcessor:
def __init__(self):
self._commands = {}
self._help_texts = {}
def command(self, name, help_text="No description"):
"""Return a decorator that registers a function as a command."""
def decorator(fn):
self._commands[name] = fn
self._help_texts[name] = help_text
return fn
return decorator
def execute(self, command_string):
"""Parse and execute a command string."""
parts = command_string.strip().split()
if not parts:
return "Empty command"
cmd_name = parts[0]
args = parts[1:]
if cmd_name == "help":
return self.get_help()
handler = self._commands.get(cmd_name)
if handler is None:
return f"Unknown command: {cmd_name}. Type 'help' for available commands."
return handler(args)
def get_help(self):
"""Return formatted help text for all registered commands."""
lines = ["Available commands:"]
for name in sorted(self._help_texts):
lines.append(f" {name:<8}- {self._help_texts[name]}")
return "\n".join(lines)
proc = CommandProcessor()
@proc.command("greet", help_text="Greet a user by name")
def greet_cmd(args):
name = args[0] if args else "World"
return f"Hello, {name}!"
@proc.command("add", help_text="Add numbers together")
def add_cmd(args):
nums = [int(x) for x in args]
return f"Sum: {sum(nums)}"
@proc.command("repeat", help_text="Repeat a word N times")
def repeat_cmd(args):
if len(args) < 2:
return "Usage: repeat <word> <count>"
word, count = args[0], int(args[1])
return f"Result: {' '.join([word] * count)}"
@proc.command("upper", help_text="Convert text to uppercase")
def upper_cmd(args):
return f"Result: {' '.join(args).upper()}"
test_commands = [
"greet Alice",
"add 10 20 30",
"repeat hello 3",
"upper hello world",
"greet",
"unknown foo",
"help",
]
for cmd_str in test_commands:
result = proc.execute(cmd_str)
print(f"> {cmd_str}")
print(f" {result}")Solution
class CommandProcessor:
def __init__(self):
self._commands = {}
self._help_texts = {}
def command(self, name, help_text="No description"):
def decorator(fn):
self._commands[name] = fn
self._help_texts[name] = help_text
return fn
return decorator
def execute(self, command_string):
parts = command_string.strip().split()
if not parts:
return "Empty command"
cmd_name = parts[0]
args = parts[1:]
if cmd_name == "help":
return self.get_help()
handler = self._commands.get(cmd_name)
if handler is None:
return f"Unknown command: {cmd_name}. Type 'help' for available commands."
return handler(args)
def get_help(self):
lines = ["Available commands:"]
for name in sorted(self._help_texts):
lines.append(f" {name:<8}- {self._help_texts[name]}")
return "\n".join(lines)
Output:
> greet Alice
Hello, Alice!
> add 10 20 30
Sum: 60
> repeat hello 3
Result: hello hello hello
> upper hello world
Result: HELLO WORLD
> greet
Hello, World!
> unknown foo
Unknown command: unknown. Type 'help' for available commands.
> help
Available commands:
add - Add numbers together
greet - Greet a user by name
repeat - Repeat a word N times
upper - Convert text to uppercase
The decorator pattern here: @proc.command("greet", ...) calls proc.command("greet", ...) which returns decorator. Python then calls decorator(greet_cmd), which registers the function and returns it unchanged. This is a parameterized decorator — a function that returns a decorator.
Why this is powerful: Commands are self-registering. Each function declares its own name and help text right where it is defined. There is no central "register all commands" function to maintain. Adding a new command is a single function definition with a decorator — zero changes to the processor.
Real-world usage: This exact pattern is used by Flask (@app.route), Click (@click.command), Discord.py (@bot.command), and many CLI frameworks. The decorator-based registry is one of Python's most elegant patterns for building extensible systems.
class CommandProcessor:
"""A plugin-based command processor where commands self-register."""
def __init__(self):
self._commands = {}
self._help_texts = {}
def command(self, name, help_text="No description"):
"""Return a decorator that registers a function as a command."""
# TODO: implement decorator that stores function and help text
pass
def execute(self, command_string):
"""Parse and execute a command string like 'command arg1 arg2'.
Return the command's result string.
Return 'Unknown command: <name>. Type help for available commands.' for unknown.
"""
# TODO: Split command string, dispatch to registered handler
pass
def get_help(self):
"""Return formatted help text for all registered commands."""
# TODO: Return sorted list of commands with their help text
pass
# Build processor
proc = CommandProcessor()
@proc.command("greet", help_text="Greet a user by name")
def greet_cmd(args):
name = args[0] if args else "World"
return f"Hello, {name}!"
@proc.command("add", help_text="Add numbers together")
def add_cmd(args):
nums = [int(x) for x in args]
return f"Sum: {sum(nums)}"
@proc.command("repeat", help_text="Repeat a word N times")
def repeat_cmd(args):
if len(args) < 2:
return "Usage: repeat <word> <count>"
word, count = args[0], int(args[1])
return f"Result: {' '.join([word] * count)}"
@proc.command("upper", help_text="Convert text to uppercase")
def upper_cmd(args):
return f"Result: {' '.join(args).upper()}"
# Test
test_commands = [
"greet Alice",
"add 10 20 30",
"repeat hello 3",
"upper hello world",
"greet",
"unknown foo",
"help",
]
for cmd_str in test_commands:
result = proc.execute(cmd_str)
print(f"> {cmd_str}")
print(f" {result}")Expected Output
> greet Alice\n Hello, Alice!\n> add 10 20 30\n Sum: 60\n> repeat hello 3\n Result: hello hello hello\n> upper hello world\n Result: HELLO WORLD\n> greet\n Hello, World!\n> unknown foo\n Unknown command: unknown. Type 'help' for available commands.\n> help\n Available commands:\n add - Add numbers together\n greet - Greet a user by name\n repeat - Repeat a word N times\n upper - Convert text to uppercaseHints
Hint 1: The command() method should return a decorator — a function that takes a function and registers it, then returns it unchanged.
Hint 2: Split the command string with .split() to get the command name and arguments list.
Hint 3: Handle "help" as a special case in execute(), or register it as a built-in command.
Build a pricing engine where discount and surcharge rules are composable functions. Each rule has a condition (should it apply?) and a modifier (how does it change the price?). Rules chain sequentially — the output of one feeds into the next.
class PricingEngine:
def __init__(self, base_price):
self.base_price = base_price
self._rules = []
def add_rule(self, name, condition_fn, modifier_fn):
self._rules.append((name, condition_fn, modifier_fn))
def calculate(self, context):
price = self.base_price
applied = []
for name, condition_fn, modifier_fn in self._rules:
if condition_fn(context):
price = modifier_fn(price, context)
applied.append(name)
return {"final_price": price, "applied_rules": applied}
engine = PricingEngine(100.00)
engine.add_rule(
"member_discount",
lambda ctx: ctx.get("is_member", False),
lambda price, ctx: price * 0.90
)
engine.add_rule(
"bulk_discount",
lambda ctx: ctx.get("quantity", 1) >= 10,
lambda price, ctx: price * 0.85
)
engine.add_rule(
"holiday_surcharge",
lambda ctx: ctx.get("is_holiday", False),
lambda price, ctx: price * 1.25
)
engine.add_rule(
"coupon",
lambda ctx: "coupon_amount" in ctx,
lambda price, ctx: max(price - ctx["coupon_amount"], 0)
)
engine.add_rule(
"minimum_price",
lambda ctx: True,
lambda price, ctx: max(price, 10.00)
)
scenarios = [
{"label": "Regular customer", "ctx": {}},
{"label": "Member", "ctx": {"is_member": True}},
{"label": "Member + bulk", "ctx": {"is_member": True, "quantity": 15}},
{"label": "Holiday + member", "ctx": {"is_member": True, "is_holiday": True}},
{"label": "Big coupon", "ctx": {"coupon_amount": 95}},
{"label": "Huge coupon (floor)", "ctx": {"coupon_amount": 200}},
]
for scenario in scenarios:
result = engine.calculate(scenario["ctx"])
applied = ", ".join(result["applied_rules"]) if result["applied_rules"] else "none"
print(f"{scenario['label']:>22}: \${result['final_price']:.2f} (rules: {applied})")Solution
class PricingEngine:
def __init__(self, base_price):
self.base_price = base_price
self._rules = []
def add_rule(self, name, condition_fn, modifier_fn):
self._rules.append((name, condition_fn, modifier_fn))
def calculate(self, context):
price = self.base_price
applied = []
for name, condition_fn, modifier_fn in self._rules:
if condition_fn(context):
price = modifier_fn(price, context)
applied.append(name)
return {"final_price": price, "applied_rules": applied}
Output:
Regular customer: $100.00 (rules: minimum_price)
Member: $90.00 (rules: member_discount, minimum_price)
Member + bulk: $76.50 (rules: member_discount, bulk_discount, minimum_price)
Holiday + member: $112.50 (rules: member_discount, holiday_surcharge, minimum_price)
Big coupon: $10.00 (rules: coupon, minimum_price)
Huge coupon (floor): $10.00 (rules: coupon, minimum_price)
Why composable rules matter: In a real pricing system, business rules change constantly — "add a 5% loyalty bonus", "remove the holiday surcharge", "add a regional tax". With composable rules, each change is adding or removing a single rule. No existing code is modified.
Rule ordering is critical: The member discount applies before the bulk discount, so bulk discount is calculated on the already-reduced price (100 * 0.90 = 90, then 90 * 0.85 = 76.50). The minimum price rule is last as a safety net — it guarantees the price never drops below $10 regardless of what discounts or coupons were applied.
Key insight: This is a pipeline pattern — each rule transforms the price and passes the result forward. The condition function decides if a rule fires, and the modifier function decides how the price changes. This separation makes rules independently testable and trivially composable.
class PricingEngine:
"""A pricing engine that applies composable discount/markup rules."""
def __init__(self, base_price):
self.base_price = base_price
self._rules = []
def add_rule(self, name, condition_fn, modifier_fn):
"""Add a pricing rule.
condition_fn(context) -> bool: whether this rule applies
modifier_fn(price, context) -> new_price: how to modify the price
"""
# TODO: Store the rule
pass
def calculate(self, context):
"""Apply all matching rules in order and return final price + breakdown.
Returns dict with 'final_price' and 'applied_rules' list."""
# TODO: Start with base_price, apply matching rules, track breakdown
passExpected Output
Regular customer: $100.00 (rules: minimum_price)\n Member: $90.00 (rules: member_discount, minimum_price)\n Member + bulk: $76.50 (rules: member_discount, bulk_discount, minimum_price)\n Holiday + member: $112.50 (rules: member_discount, holiday_surcharge, minimum_price)\n Big coupon: $10.00 (rules: coupon, minimum_price)\n Huge coupon (floor): $10.00 (rules: coupon, minimum_price)Hints
Hint 1: Store rules as a list of tuples: (name, condition_fn, modifier_fn). Order matters — rules apply sequentially.
Hint 2: In calculate(), iterate through rules: if condition_fn(context) is True, apply modifier_fn(price, context) and record the rule name.
Hint 3: The minimum_price rule always applies (condition returns True) and enforces a price floor.
Build an event bus where handlers register for named events with priorities. When an event is emitted, handlers run in priority order and can modify the event data flowing through. This combines observer pattern, dispatch tables, and chain of responsibility.
class EventBus:
def __init__(self):
self._listeners = {}
self._log = []
def on(self, event_name, handler, priority=0):
"""Register a handler for an event. Lower priority = runs first."""
if event_name not in self._listeners:
self._listeners[event_name] = []
self._listeners[event_name].append((priority, handler))
def emit(self, event_name, data=None):
"""Emit an event, calling handlers in priority order."""
if data is None:
data = {}
handlers = self._listeners.get(event_name, [])
sorted_handlers = sorted(handlers, key=lambda x: x[0])
self._log.append(f"{event_name} -> {len(sorted_handlers)} handlers")
for priority, handler in sorted_handlers:
result = handler(event_name, data)
if result is not None and isinstance(result, dict):
data.update(result)
return data
def get_log(self):
return list(self._log)
bus = EventBus()
# User registration pipeline
bus.on("user:register", lambda e, d: (
None if "@" in d.get("email", "")
else {"errors": d.get("errors", []) + ["Invalid email"]}
), priority=1)
bus.on("user:register", lambda e, d: (
None if len(d.get("password", "")) >= 8
else {"errors": d.get("errors", []) + ["Password too short"]}
), priority=2)
bus.on("user:register", lambda e, d: (
{"username": d["email"].split("@")[0]} if not d.get("errors") and "email" in d
else None
), priority=3)
bus.on("user:register", lambda e, d: (
{"status": "created", "message": f"Welcome {d.get('username', 'user')}!"}
if not d.get("errors")
else {"status": "failed", "message": f"Registration failed: {', '.join(d['errors'])}"}
), priority=10)
# Order processing pipeline
bus.on("order:process", lambda e, d: (
{"discount": 0.1} if d.get("is_member") else None
), priority=1)
bus.on("order:process", lambda e, d: (
{"total": d["price"] * (1 - d.get("discount", 0))}
), priority=5)
bus.on("order:process", lambda e, d: (
{"receipt": f"Order #{d.get('order_id')}: \${d['total']:.2f}"}
), priority=10)
# Test events
print("=== User Registration ===")
result1 = bus.emit("user:register", {"email": "[email protected]", "password": "secure123"})
print(f" Status: {result1.get('status')} | {result1.get('message')}")
result2 = bus.emit("user:register", {"email": "bad-email", "password": "short"})
print(f" Status: {result2.get('status')} | {result2.get('message')}")
result3 = bus.emit("user:register", {"email": "[email protected]", "password": "short"})
print(f" Status: {result3.get('status')} | {result3.get('message')}")
print("\n=== Order Processing ===")
result4 = bus.emit("order:process", {"order_id": 1001, "price": 50.00, "is_member": True})
print(f" {result4.get('receipt')}")
result5 = bus.emit("order:process", {"order_id": 1002, "price": 75.00, "is_member": False})
print(f" {result5.get('receipt')}")
print(f"\n=== Event Log ({len(bus.get_log())} events) ===")
for entry in bus.get_log():
print(f" {entry}")Solution
class EventBus:
def __init__(self):
self._listeners = {}
self._log = []
def on(self, event_name, handler, priority=0):
if event_name not in self._listeners:
self._listeners[event_name] = []
self._listeners[event_name].append((priority, handler))
def emit(self, event_name, data=None):
if data is None:
data = {}
handlers = self._listeners.get(event_name, [])
sorted_handlers = sorted(handlers, key=lambda x: x[0])
self._log.append(f"{event_name} -> {len(sorted_handlers)} handlers")
for priority, handler in sorted_handlers:
result = handler(event_name, data)
if result is not None and isinstance(result, dict):
data.update(result)
return data
def get_log(self):
return list(self._log)
Output:
=== User Registration ===
Status: created | Welcome alice!
Status: failed | Registration failed: Invalid email, Password too short
Status: failed | Registration failed: Password too short
=== Order Processing ===
Order #1001: $45.00
Order #1002: $75.00
=== Event Log (5 events) ===
user:register -> 4 handlers
user:register -> 4 handlers
user:register -> 4 handlers
order:process -> 3 handlers
order:process -> 3 handlers
Three patterns combined:
-
Observer/Pub-Sub: Handlers subscribe to named events without knowing who emits them. The bus decouples producers from consumers.
-
Chain of Responsibility: Handlers run in priority order, and each can modify the data flowing through. Early handlers (validation at priority 1-2) can set error flags that later handlers (at priority 10) check before proceeding.
-
Dispatch Table: The
_listenersdict maps event names to handler lists, providing O(1) lookup for which handlers to invoke.
Why data.update(result) is the key design choice: Handlers return partial dicts that get merged into the flowing data. This means each handler only needs to specify what it changes — it does not need to reconstruct the entire data object. The validation handler adds "errors", the username handler adds "username", and the final handler reads both to decide the outcome.
Real-world usage: This pattern is the foundation of event-driven architectures — Node.js EventEmitter, Python's blinker library, Django signals, game engines, and GUI frameworks all use variations of this pub-sub-with-data-flow pattern.
class EventBus:
"""An event-driven system where listeners register for specific events."""
def __init__(self):
self._listeners = {} # event_name -> list of (priority, handler)
self._log = []
def on(self, event_name, handler, priority=0):
"""Register a handler for an event. Lower priority number = runs first."""
# TODO: Store handler with priority
pass
def emit(self, event_name, data=None):
"""Emit an event — call all registered handlers in priority order.
Each handler receives (event_name, data) and can return a dict to
modify data for subsequent handlers, or None to pass through.
Return the final data after all handlers."""
# TODO: Get handlers, sort by priority, call each, merge returned dicts
pass
def get_log(self):
"""Return the event log."""
return list(self._log)Expected Output
=== User Registration ===
Status: created | Welcome alice!
Status: failed | Registration failed: Invalid email, Password too short
Status: failed | Registration failed: Password too short
=== Order Processing ===
Order #1001: $45.00
Order #1002: $75.00
=== Event Log (5 events) ===
user:register -> 4 handlers
user:register -> 4 handlers
user:register -> 4 handlers
order:process -> 3 handlers
order:process -> 3 handlersHints
Hint 1: Sort handlers by priority before calling them — use sorted() with a key function on the priority value.
Hint 2: When a handler returns a dict, merge it into the data with data.update(result). If it returns None, keep data unchanged.
Hint 3: The log should record event names and handler counts, not the data itself.