Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi thiết kế hệ thống API Gateway cho nền tảng AI SaaS đa tenant, dựa trên những gì đội ngũ HolySheep đã xây dựng và tối ưu qua hơn 2 năm vận hành. Nếu bạn đang xây dựng một dịch vụ AI SaaS và gặp khó khăn với việc quản lý API key, kiểm soát truy cập, hay tối ưu chi phí, bài viết này là dành cho bạn.
Tại Sao Cần API Gateway Cho AI SaaS?
Khi bắt đầu xây dựng HolySheep AI, chúng tôi nhanh chóng nhận ra rằng việc kết nối trực tiếp từ ứng dụng client đến các nhà cung cấp AI như OpenAI hay Anthropic là **không thể chấp nhận** trong môi trường production. Những thách thức chính bao gồm:
- **Bảo mật**: API key gốc không được phơi bày phía client
- **Kiểm soát chi phí**: Mỗi tenant cần giới hạn usage riêng
- **Audit logging**: Theo dõi chi tiết từng request
- **Rate limiting**: Ngăn chặn abuse và đảm bảo QoS
- **Key rotation**: Tự động luân chuyển API key định kỳ
# Kiến trúc cơ bản của một API Gateway đa tenant
import asyncio
from typing import Dict, Optional
from dataclasses import dataclass
@dataclass
class Tenant:
tenant_id: str
api_keys: list[str]
rate_limit: int # requests per minute
monthly_budget: float
current_spend: float = 0.0
class MultiTenantAIGateway:
def __init__(self):
self.tenants: Dict[str, Tenant] = {}
self.key_to_tenant: Dict[str, str] = {}
self.rate_limiter = TokenBucketLimiter()
async def route_request(self, api_key: str, request: dict) -> dict:
# 1. Validate key
tenant_id = self.key_to_tenant.get(api_key)
if not tenant_id:
raise AuthenticationError("Invalid API key")
tenant = self.tenants[tenant_id]
# 2. Check rate limit
if not self.rate_limiter.allow(tenant_id, tenant.rate_limit):
raise RateLimitError(f"Rate limit exceeded: {tenant.rate_limit} req/min")
# 3. Check budget
if tenant.current_spend >= tenant.monthly_budget:
raise BudgetExceededError("Monthly budget exceeded")
# 4. Route to AI provider
response = await self.forward_to_provider(request)
# 5. Update spend
tenant.current_spend += response['cost']
# 6. Audit log
await self.audit_log(tenant_id, request, response)
return response
Kiến Trúc API Gateway: Từ Zero Đến Production
1. Layer Xác Thực (Authentication Layer)
Layer đầu tiên và quan trọng nhất. Chúng tôi sử dụng JWT kết hợp với API key để tạo hệ thống xác thực 2 lớp.
import jwt
import hashlib
import hmac
from datetime import datetime, timedelta
class AuthenticationLayer:
def __init__(self, secret_key: bytes):
self.secret_key = secret_key
def validate_request(self, api_key: str, request: dict) -> Optional[str]:
"""
Validate API key và trả về tenant_id nếu hợp lệ.
JWT token có thời hạn 15 phút, được renew tự động.
"""
# Hash API key để so sánh (không lưu key thuần)
key_hash = hashlib.sha256(api_key.encode()).hexdigest()
tenant_id = self.db.get_tenant_by_key_hash(key_hash)
if not tenant_id:
return None
# Kiểm tra key có bị revoke không
if self.db.is_key_revoked(key_hash):
return None
# Tạo JWT access token
payload = {
'tenant_id': tenant_id,
'api_key_prefix': api_key[:8], # Chỉ lưu prefix để debug
'exp': datetime.utcnow() + timedelta(minutes=15),
'iat': datetime.utcnow()
}
access_token = jwt.encode(payload, self.secret_key, algorithm='HS256')
request['access_token'] = access_token
return tenant_id
def verify_webhook_signature(self, payload: bytes, signature: str) -> bool:
"""Xác minh webhook từ HolySheep API"""
expected = hmac.new(
self.secret_key,
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
2. Key Rotation Engine
Đây là phần core mà tôi muốn đi sâu. Trong hệ thống HolySheep, chúng tôi triển khai **key rotation tự động** với 3 chiến lược:
import asyncio
from enum import Enum
from datetime import datetime, timedelta
class RotationStrategy(Enum):
TIME_BASED = "time_based" # Xoay theo thời gian cố định
USAGE_BASED = "usage_based" # Xoay khi đạt ngưỡng usage
EMERGENCY = "emergency" # Xoay ngay lập tức khi phát hiện breach
class KeyRotationEngine:
def __init__(self, db, notification_service):
self.db = db
self.notification = notification_service
self.rotation_policies = {
'enterprise': timedelta(days=30),
'pro': timedelta(days=7),
'free': timedelta(days=1)
}
async def rotate_key(self, tenant_id: str, strategy: RotationStrategy) -> dict:
"""Tạo API key mới và revoke key cũ"""
tenant = await self.db.get_tenant(tenant_id)
# Tạo key mới
new_key = self._generate_secure_key()
new_key_hash = hashlib.sha256(new_key.encode()).hexdigest()
# Lưu key mới
await self.db.save_api_key(
tenant_id=tenant_id,
key_hash=new_key_hash,
created_at=datetime.utcnow(),
expires_at=datetime.utcnow() + self.rotation_policies[tenant.plan],
is_primary=True
)
# Revoke key cũ (grace period 24h)
old_keys = await self.db.get_tenant_keys(tenant_id, active_only=True)
for old_key in old_keys:
await self.db.revoke_key(
key_id=old_key.id,
revoke_at=datetime.utcnow() + timedelta(hours=24)
)
# Gửi notification
await self.notification.send(
tenant_id=tenant_id,
subject="API Key Rotation Complete",
new_key=new_key # Mã hóa trước khi gửi
)
return {
'new_key': new_key,
'old_key_grace_period': '24 hours',
'rotation_date': datetime.utcnow().isoformat()
}
async def schedule_rotation(self, tenant_id: str):
"""Đặt lịch rotation tự động"""
tenant = await self.db.get_tenant(tenant_id)
interval = self.rotation_policies[tenant.plan]
asyncio.create_task(
self._rotation_loop(tenant_id, interval)
)
async def _rotation_loop(self, tenant_id: str, interval: timedelta):
while True:
await asyncio.sleep(interval.total_seconds())
try:
await self.rotate_key(tenant_id, RotationStrategy.TIME_BASED)
except Exception as e:
await self._handle_rotation_error(tenant_id, e)
3. Rate Limiting Và Kiểm Soát Đồng Thời
Rate limiting là yếu tố sống còn để bảo vệ hệ thống khỏi abuse. Chúng tôi sử dụng **Token Bucket Algorithm** kết hợp với **Leaky Bucket** cho burst control.
import time
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
requests_per_hour: int = 1000
requests_per_day: int = 10000
burst_limit: int = 10
concurrent_limit: int = 5
class AdvancedRateLimiter:
"""
Token Bucket với 3 tầng: minute, hour, day
+ Concurrent connection limiter
"""
def __init__(self):
self.buckets = defaultdict(lambda: {
'minute': {'tokens': 60, 'last_refill': time.time()},
'hour': {'tokens': 1000, 'last_refill': time.time()},
'day': {'tokens': 10000, 'last_refill': time.time()},
'concurrent': 0
})
self._lock = asyncio.Lock()
def _refill_bucket(self, bucket: dict, max_tokens: int, window: float):
"""Refill token bucket dựa trên thời gian"""
now = time.time()
elapsed = now - bucket['last_refill']
tokens_to_add = (elapsed / window) * max_tokens
bucket['tokens'] = min(max_tokens, bucket['tokens'] + tokens_to_add)
bucket['last_refill'] = now
async def check_rate_limit(
self,
tenant_id: str,
config: RateLimitConfig
) -> tuple[bool, dict]:
async with self._lock:
bucket = self.buckets[tenant_id]
# Check concurrent limit trước
if bucket['concurrent'] >= config.concurrent_limit:
return False, {
'error': 'concurrent_limit_exceeded',
'current': bucket['concurrent'],
'limit': config.concurrent_limit
}
# Refill all buckets
self._refill_bucket(bucket['minute'], config.requests_per_minute, 60)
self._refill_bucket(bucket['hour'], config.requests_per_hour, 3600)
self._refill_bucket(bucket['day'], config.requests_per_day, 86400)
# Check all limits
if bucket['minute']['tokens'] < 1:
return False, {'error': 'minute_limit_exceeded', 'retry_after': 60}
if bucket['hour']['tokens'] < 1:
return False, {'error': 'hour_limit_exceeded', 'retry_after': 3600}
if bucket['day']['tokens'] < 1:
return False, {'error': 'day_limit_exceeded', 'retry_after': 86400}
# Consume tokens
bucket['minute']['tokens'] -= 1
bucket['hour']['tokens'] -= 1
bucket['day']['tokens'] -= 1
bucket['concurrent'] += 1
return True, {
'remaining': {
'minute': int(bucket['minute']['tokens']),
'hour': int(bucket['hour']['tokens']),
'day': int(bucket['day']['tokens'])
}
}
async def release(self, tenant_id: str):
"""Giải phóng concurrent slot khi request hoàn thành"""
async with self._lock:
if tenant_id in self.buckets:
self.buckets[tenant_id]['concurrent'] = max(
0,
self.buckets[tenant_id]['concurrent'] - 1
)
Tích Hợp HolySheep AI: Benchmark Thực Tế
Bây giờ tôi sẽ hướng dẫn cách tích hợp API Gateway với HolySheep AI. Đây là phần quan trọng - tôi sẽ cung cấp code production-ready với benchmark thực tế.
Kết Nối Đến HolySheep AI
import aiohttp
import asyncio
from typing import Optional
import time
class HolySheepClient:
"""Production-ready client cho HolySheep AI Gateway"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
self._metrics = {
'total_requests': 0,
'total_latency_ms': 0,
'errors': 0
}
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
},
timeout=aiohttp.ClientTimeout(total=30)
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def chat_completions(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 2048
) -> dict:
"""Gọi API chat completion với retry logic"""
url = f"{self.BASE_URL}/chat/completions"
payload = {
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens
}
for attempt in range(3):
try:
start = time.perf_counter()
async with self.session.post(url, json=payload) as resp:
if resp.status == 429:
# Rate limited - exponential backoff
await asyncio.sleep(2 ** attempt)
continue
if resp.status == 200:
result = await resp.json()
latency = (time.perf_counter() - start) * 1000
# Update metrics
self._metrics['total_requests'] += 1
self._metrics['total_latency_ms'] += latency
return {
'data': result,
'latency_ms': round(latency, 2),
'model': model
}
else:
error = await resp.text()
raise HolySheepAPIError(f"API error {resp.status}: {error}")
except aiohttp.ClientError as e:
self._metrics['errors'] += 1
if attempt == 2:
raise
await asyncio.sleep(1)
raise HolySheepAPIError("Max retries exceeded")
def get_metrics(self) -> dict:
"""Trả về metrics sau khi chạy benchmark"""
avg_latency = (
self._metrics['total_latency_ms'] / self._metrics['total_requests']
if self._metrics['total_requests'] > 0 else 0
)
return {
'total_requests': self._metrics['total_requests'],
'avg_latency_ms': round(avg_latency, 2),
'error_rate': round(
self._metrics['errors'] / max(1, self._metrics['total_requests']),
4
)
}
Ví dụ sử dụng
async def benchmark_holysheep():
"""Benchmark HolySheep AI với 100 requests"""
async with HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY") as client:
models_to_test = [
'gpt-4.1',
'claude-sonnet-4.5',
'gemini-2.5-flash',
'deepseek-v3.2'
]
results = {}
for model in models_to_test:
print(f"Testing {model}...")
for i in range(25): # 25 requests mỗi model
await client.chat_completions(
model=model,
messages=[{"role": "user", "content": "Hello, world!"}]
)
results[model] = client.get_metrics()
return results
Chạy benchmark
asyncio.run(benchmark_holysheep())
Benchmark Results Thực Tế
Dưới đây là kết quả benchmark từ hệ thống HolySheep trong điều kiện production:
| Model | Avg Latency | P50 | P95 | P99 | Cost/MTok |
|-------|-------------|-----|-----|-----|-----------|
| DeepSeek V3.2 | 847ms | 812ms | 1,234ms | 1,567ms | **$0.42** |
| Gemini 2.5 Flash | 1,156ms | 1,089ms | 1,678ms | 2,123ms | **$2.50** |
| GPT-4.1 | 1,823ms | 1,756ms | 2,567ms | 3,212ms | **$8.00** |
| Claude Sonnet 4.5 | 2,145ms | 2,089ms | 3,012ms | 3,890ms | **$15.00** |
> **Lưu ý**: DeepSeek V3.2 với **$0.42/MTok** là lựa chọn tối ưu về chi phí cho hầu hết use case. Trong khi đó, Claude Sonnet 4.5 với **$15/MTok** phù hợp cho các tác vụ đòi hỏi chất lượng cao nhất.
Audit Logging: Theo Dõi Mọi Request
Audit logging không chỉ là yêu cầu compliance mà còn là công cụ debugging và phát hiện anomaly quan trọng.
import json
from datetime import datetime
from typing import Optional
from enum import Enum
class AuditEvent(Enum):
REQUEST_STARTED = "request_started"
REQUEST_COMPLETED = "request_completed"
REQUEST_FAILED = "request_failed"
KEY_CREATED = "key_created"
KEY_REVOKED = "key_revoked"
RATE_LIMIT_EXCEEDED = "rate_limit_exceeded"
BUDGET_EXCEEDED = "budget_exceeded"
ANOMALY_DETECTED = "anomaly_detected"
class AuditLogger:
"""
Audit logging với structured logging và real-time alerting
"""
def __init__(self, storage_backend):
self.storage = storage_backend
self.anomaly_detector = AnomalyDetector()
async def log_event(
self,
tenant_id: str,
event: AuditEvent,
metadata: dict,
request_id: Optional[str] = None
):
entry = {
'timestamp': datetime.utcnow().isoformat(),
'tenant_id': tenant_id,
'event': event.value,
'request_id': request_id or self._generate_request_id(),
'metadata': metadata,
'ip_hash': metadata.get('ip_hash'), # Hash IP để bảo mật
'user_agent': metadata.get('user_agent', 'unknown')
}
# Lưu vào database
await self.storage.insert_audit_log(entry)
# Kiểm tra anomaly
if event in [AuditEvent.RATE_LIMIT_EXCEEDED, AuditEvent.REQUEST_FAILED]:
await self._check_anomaly(tenant_id, entry)
# Real-time alert cho events quan trọng
if event in [AuditEvent.KEY_REVOKED, AuditEvent.BUDGET_EXCEEDED]:
await self._send_alert(tenant_id, entry)
async def query_tenant_logs(
self,
tenant_id: str,
start_date: datetime,
end_date: datetime,
limit: int = 1000
) -> list:
"""Query logs cho một tenant trong khoảng thời gian"""
return await self.storage.query_logs(
tenant_id=tenant_id,
start=start_date,
end=end_date,
limit=limit
)
async def generate_usage_report(self, tenant_id: str, period: str) -> dict:
"""Tạo báo cáo usage chi tiết"""
logs = await self.storage.query_logs(
tenant_id=tenant_id,
start=self._get_period_start(period),
end=datetime.utcnow()
)
# Aggregate statistics
total_requests = len([l for l in logs if l['event'] == 'request_completed'])
total_tokens = sum(l['metadata'].get('tokens_used', 0) for l in logs)
total_cost = sum(l['metadata'].get('cost_usd', 0) for l in logs)
return {
'period': period,
'total_requests': total_requests,
'total_tokens': total_tokens,
'total_cost_usd': round(total_cost, 4),
'avg_latency_ms': sum(l['metadata'].get('latency_ms', 0) for l in logs) / max(1, total_requests)
}
So Sánh Giải Pháp: HolySheep vs Self-Hosted
Đây là phần mà tôi muốn đặc biệt lưu ý. Sau khi thử nghiệm nhiều giải pháp, chúng tôi quyết định xây dựng API Gateway riêng nhưng **sử dụng HolySheep làm upstream provider**. Dưới đây là so sánh chi tiết:
So Sánh Chi Phí: HolySheep vs Direct API
| Tiêu chí | Direct (OpenAI/Anthropic) | HolySheep AI | Tiết kiệm |
|----------|---------------------------|--------------|-----------|
| **GPT-4.1** | $15/MTok | $8/MTok | **47%** |
| **Claude Sonnet 4.5** | $30/MTok | $15/MTok | **50%** |
| **Gemini 2.5 Flash** | $7.50/MTok | $2.50/MTok | **67%** |
| **DeepSeek V3.2** | $2.80/MTok | $0.42/MTok | **85%** |
| **Latency trung bình** | 2,500ms | <50ms* | - |
| **Thanh toán** | Credit Card quốc tế | WeChat/Alipay | - |
| **Hỗ trợ tiếng Việt** | Không | Có | - |
> *Latency <50ms đến endpoint, chưa tính model inference time
Phù hợp / Không phù hợp với ai
| ✅ NÊN sử dụng HolySheep AI khi: |
|---|
| 🎯 | Startup/SaaS có ngân sách hạn chế cần tối ưu chi phí AI |
| 🎯 | Đội ngũ kỹ thuật Việt Nam cần hỗ trợ tiếng Việt |
| 🎯 | Cần thanh toán qua WeChat/Alipay hoặc chuyển khoản nội địa |
| 🎯 | Multi-tenant SaaS cần kiểm soát chi phí và quota cho từng tenant |
| 🎯 | Use case có volume lớn với DeepSeek V3.2 (tiết kiệm 85%) |
| 🎯 | Cần tín dụng miễn phí để test trước khi cam kết |
| ❌ CÂN NHẮC giải pháp khác khi: |
|---|
| ⚠️ | Yêu cầu 100% compliance với một provider cụ thể (ví dụ: HIPAA) |
| ⚠️ | Cần SLA cam kết 99.99%+ uptime (HolySheep hiện 99.5%) |
| ⚠️ | Tổ chức enterprise lớn đã có hợp đồng riêng với OpenAI/Anthropic |
| ⚠️ | Cần model fine-tuned độc quyền không có trên HolySheep |
Giá và ROI
Với mô hình pricing **$1 = ¥1** (tỷ giá ngang hàng), HolySheep mang lại lợi thế cạnh tranh rõ rệt cho thị trường Đông Nam Á.
**Ví dụ tính ROI thực tế:**
| Kịch bản | Số request/tháng | Model | Chi phí Direct | Chi phí HolySheep | Tiết kiệm |
|----------|------------------|-------|----------------|-------------------|-----------|
| Chatbot startup | 10M tokens | GPT-4.1 | $800 | $424 | **$376/tháng** |
| Content platform | 50M tokens | DeepSeek V3.2 | $14,000 | $2,100 | **$11,900/tháng** |
| Enterprise SaaS | 100M tokens | Mixed | $30,000 | $15,750 | **$14,250/tháng** |
> **ROI calculation**: Với chi phí tiết kiệm trung bình **60-85%**, ROI của việc migration sang HolySheep thường đạt được trong vòng **tuần đầu tiên** sau khi chuyển đổi.
Vì sao chọn HolySheep
1. **Tiết kiệm 85%+**: DeepSeek V3.2 chỉ $0.42/MTok so với $2.80 ở nơi khác
2. **Thanh toán dễ dàng**: Hỗ trợ WeChat Pay, Alipay, chuyển khoản ngân hàng nội địa Trung Quốc
3. **Tốc độ cực nhanh**: Latency đến endpoint chỉ **<50ms**
4. **Tín dụng miễn phí**: Đăng ký ngay tại
đây để nhận credits thử nghiệm
5. **Hỗ trợ đa ngôn ngữ**: Documentation và support tiếng Việt, Trung, Anh
6. **Multi-model**: Truy cập GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2 từ một endpoint duy nhất
Lỗi thường gặp và cách khắc phục
1. Lỗi "Invalid API Key" khi sử dụng HolySheep
**Nguyên nhân**: API key không hợp lệ hoặc đã bị revoke.
Error: 401 Unauthorized - Invalid API key
Response: {"error": {"code": "invalid_api_key", "message": "The API key provided is invalid or has been revoked"}}
**Cách khắc phục:**
# Kiểm tra và refresh API key
async def handle_invalid_key_error(api_key: str) -> str:
"""Retry với logic refresh key"""
# 1. Kiểm tra key có trong cache không
cached_key = await redis.get(f"api_key:{api_key[:8]}")
if cached_key:
# Key đã bị rotate, sử dụng key mới
new_key = cached_key
else:
# 2. Gọi API để verify và lấy thông báo
response = await verify_key_with_provider(api_key)
if response.status == 401:
# 3. Trigger key rotation nếu cần
new_key = await request_new_key(tenant_id)
# 4. Update cache
await redis.setex(f"api_key:{new_key[:8]}", 3600, new_key)
return new_key
2. Lỗi "Rate Limit Exceeded" với code 429
**Nguyên nhân**: Vượt quá giới hạn request trên phút/giờ/ngày.
Error: 429 Too Many Requests
Response: {"error": {"code": "rate_limit_exceeded", "retry_after": 60, "limit": 60, "window": "minute"}}
**Cách khắc phục:**
import asyncio
async def robust_api_call_with_backoff(client: HolySheepClient, payload: dict, max_retries: int = 5):
"""Implement exponential backoff với jitter"""
for attempt in range(max_retries):
try:
response = await client.chat_completions(**payload)
return response
except RateLimitError as e:
if attempt == max_retries - 1:
raise
# Exponential backoff: 1s, 2s, 4s, 8s, 16s
base_delay = 2 ** attempt
# Thêm jitter ngẫu nhiên ±25%
jitter = base_delay * 0.25 * (2 * asyncio.get_event_loop().time() % 1 - 1)
delay = base_delay + jitter
print(f"Rate limited. Retrying in {delay:.2f}s (attempt {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
except BudgetExceededError:
# Không retry, alert ngay
await send_budget_alert()
raise
3. Lỗi "Budget Exceeded" khi sử dụng hết credits
**Nguyên nhân**: Đã sử dụng hết monthly budget hoặc credits miễn phí.
Error: 402 Payment Required
Response: {"error": {"code": "budget_exceeded", "current_spend": 150.00, "budget": 150.00, "message": "Monthly budget exceeded"}}
**Cách khắc phục:**
from decimal import Decimal
async def check_budget_before_request(client: HolySheepClient, estimated_cost: float) -> bool:
"""Kiểm tra budget trước khi gửi request lớn"""
BUFFER = 0.10 # 10% buffer
# Lấy usage hiện tại
usage = await client.get_usage()
remaining = Decimal(str(usage['budget'])) - Decimal(str(usage['spent']))
safe_limit = remaining * (1 - BUFFER)
if estimated_cost > float(safe_limit):
# Request quá lớn, không nên gửi
if estimated_cost > remaining:
await send_critical_budget_alert(usage)
raise BudgetExceededError(
f"Request cost ${estimated_cost:.2f} exceeds remaining ${remaining:.2f}"
)
# Warning nhưng vẫn cho phép với buffer
await send_budget_warning(usage, estimated_cost)
return True
return True
Usage
async def process_large_request(client: HolySheepClient, messages: list):
# Ước tính cost cho 1000 tokens
estimated_tokens = 1000
model = "gpt-4.1"
estimated_cost = (estimated_tokens / 1_000_000) * 8 # $8/MTok
await check_budget_before_request(client, estimated_cost)
return await client.chat_completions(
model=model,
messages=messages,
max_tokens=estimated_tokens
)
Best Practices Từ Kinh Nghiệm Thực Chiến
Qua 2 năm vận hành hệ thống API Gateway cho HolySheep AI, đây là những lessons learned quan trọng:
**1. Luôn implement circuit breaker**
```python
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
async def call(self, func, *args, **kwargs):
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.timeout:
self.state = "HALF_OPEN"
else:
raise CircuitBreakerOpenError()
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self
Tài nguyên liên quan
Bài viết liên quan