Build a local failure-test service in Colab, then observe real HTTP errors, timeout, retry and recovery.
แทนการพึ่ง public website เพื่อสร้าง 404/500/timeout เราสร้าง test service ของเราเองใน Colab แล้วทดลอง failure จริงผ่าน HTTP
Install dependencies, then define a local FastAPI service containing predictable failure scenarios.
/unstable deliberately returns 503 twice, then succeeds on the third request.!pip -q install fastapi uvicorn requests
print("Packages ready ✓")from fastapi import FastAPI, Response
import time
app = FastAPI(title="KASA Reliability Lab")
attempt_counter = {"unstable": 0}
@app.get("/ok")
def ok():
return {"status": "ok", "message": "Service is healthy"}
@app.get("/not-found")
def not_found():
return Response(
content='{"detail":"Resource not found"}',
status_code=404,
media_type="application/json"
)
@app.get("/server-error")
def server_error():
return Response(
content='{"detail":"Temporary server failure"}',
status_code=500,
media_type="application/json"
)
@app.get("/slow")
def slow():
time.sleep(3)
return {"status": "ok", "message": "Slow response finished"}
@app.get("/unstable")
def unstable():
attempt_counter["unstable"] += 1
if attempt_counter["unstable"] < 3:
return Response(
content='{"detail":"Temporary failure"}',
status_code=503,
media_type="application/json"
)
return {
"status": "ok",
"attempt": attempt_counter["unstable"]
}
print("Reliability test service ready ✓")เหมือน Ex2: Uvicorn ทำงานใน background thread และ notebook ไปทำ Cell ถัดไปได้ทันที
BASE_URL ถูกสร้างอัตโนมัติเป็น http://127.0.0.1:8001import threading
import time
import uvicorn
HOST = "127.0.0.1"
PORT = 8001
BASE_URL = f"http://{HOST}:{PORT}"
config = uvicorn.Config(
app,
host=HOST,
port=PORT,
log_level="warning"
)
api_server = uvicorn.Server(config)
server_thread = threading.Thread(
target=api_server.run,
daemon=True
)
server_thread.start()
for _ in range(20):
if api_server.started:
break
time.sleep(0.25)
print("Reliability API running ✓")
print("BASE_URL =", BASE_URL)raise_for_status() converts HTTP error status into a Python exception that we can handle.import requests
paths = [
"/ok",
"/not-found",
"/server-error"
]
for path in paths:
url = BASE_URL + path
try:
response = requests.get(url, timeout=2)
print("\nGET", path)
print("Status:", response.status_code)
response.raise_for_status()
print("Result: SUCCESS")
except requests.HTTPError as e:
print("Handled:", type(e).__name__)/slow deliberately waits 3 seconds. The client allows only 1 second.
requests ไม่ใช่การ print จำลองimport requests
try:
print("Calling /slow with timeout=1 second ...")
response = requests.get(
BASE_URL + "/slow",
timeout=1
)
print("Status:", response.status_code)
except requests.Timeout:
print("TIMEOUT handled ✓")
print("The client stopped waiting before the server finished.")/unstable returns HTTP 503 on attempts 1–2 and HTTP 200 on attempt 3.
import requests
import time
# Reset the demo counter so the endpoint fails twice again.
attempt_counter["unstable"] = 0
max_attempts = 4
for attempt in range(1, max_attempts + 1):
try:
print(f"Attempt {attempt}/{max_attempts}")
response = requests.get(
BASE_URL + "/unstable",
timeout=2
)
response.raise_for_status()
print("SUCCESS:", response.status_code)
print("JSON:", response.json())
break
except requests.RequestException as e:
print("FAILED:", type(e).__name__)
if attempt < max_attempts:
print("Retrying in 1 second...")
time.sleep(1)
else:
print("FAILED after retries")เปลี่ยน fixed delay เป็น 1s → 2s → 4s เพื่อไม่ยิง request ซ้ำถี่เกินไปเมื่อ service มีปัญหา
import requests
import time
attempt_counter["unstable"] = 0
max_attempts = 4
started = time.time()
for attempt in range(1, max_attempts + 1):
try:
print(f"Attempt {attempt}/{max_attempts}")
response = requests.get(
BASE_URL + "/unstable",
timeout=2
)
response.raise_for_status()
print("SUCCESS:", response.json())
break
except requests.RequestException as e:
print("FAILED:", type(e).__name__)
if attempt < max_attempts:
delay = 2 ** (attempt - 1)
print(f"Backoff: {delay}s")
time.sleep(delay)
print(f"Elapsed: {time.time() - started:.2f}s")api_server.should_exit = True
server_thread.join(timeout=5)
print("Reliability API stopped ✓")/unstable ให้ fail แบบสุ่ม แล้วเปรียบเทียบ fixed retry กับ exponential backoff ว่า behavior ต่างกันอย่างไร