#Lab: Design Patterns Implementation
Implement common cloud design patterns.
#🎯 Objectives
- Implement circuit breaker
- Set up retry with backoff
- Create health endpoints
#Task 1: Circuit Breaker (Python)
python
1import time
2
3class CircuitBreaker:
4 def __init__(self, failure_threshold=5, recovery_timeout=30):
5 self.failure_count = 0
6 self.failure_threshold = failure_threshold
7 self.recovery_timeout = recovery_timeout
8 self.last_failure_time = None
9 self.state = "CLOSED"
10
11 def call(self, func, *args, **kwargs):
12 if self.state == "OPEN":
13 if time.time() - self.last_failure_time > self.recovery_timeout:
14 self.state = "HALF_OPEN"
15 else:
16 raise Exception("Circuit is OPEN")
17
18 try:
19 result = func(*args, **kwargs)
20 self.failure_count = 0
21 self.state = "CLOSED"
22 return result
23 except Exception as e:
24 self.failure_count += 1
25 self.last_failure_time = time.time()
26 if self.failure_count >= self.failure_threshold:
27 self.state = "OPEN"
28 raise e#Task 2: Retry with Backoff
python
1import time
2import random
3
4def retry_with_backoff(func, max_retries=3, base_delay=1):
5 for attempt in range(max_retries):
6 try:
7 return func()
8 except Exception as e:
9 if attempt == max_retries - 1:
10 raise
11 delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
12 time.sleep(delay)#Task 3: Health Endpoint
python
1from flask import Flask, jsonify
2
3app = Flask(__name__)
4
5@app.route('/health')
6def health():
7 return jsonify({
8 "status": "healthy",
9 "checks": {
10 "database": check_database(),
11 "cache": check_cache()
12 }
13 })#✅ Success Criteria
- Circuit breaker prevents cascade
- Retries succeed after failures
- Health endpoint reports status