怎样用脚本降级API服务？

本文目录导读：

1. 场景一：功能降级/熔断（推荐方式）
2. 场景二：API版本降级（版本回退）
3. 更高级的降级策略（生产环境常用）

“降级API服务”通常有两种含义，需要先明确你的具体场景：

1. 脚本降级（功能降级 / 熔断）：当上游服务或依赖不可用时，API 自动返回默认值或缓存数据，而不是报错，这是容错设计的一部分。
2. API版本降级：调用新版本的API时，发现不支持，自动退回到旧版本的API接口。

下面分别给出两种场景的脚本示例和实现思路。

功能降级/熔断（推荐方式）

这是最常见的降级场景,假设你的API依赖一个外部天气服务，当天气服务超时或报错时，脚本不再等待或传播错误，而是立即返回一个缓存数据或默认值。

示例脚本（Python + 简单重试/降级逻辑）

import requests
import time
import json
# 缓存数据（模拟）
CACHE_DATA = {"weather": "sunny", "temp": 25}
def fetch_weather(city):
    """
    调用外部API，失败则降级到缓存
    """
    try:
        # 设置超时（防止长时间挂起）
        response = requests.get(
            f"https://api.weather.com/v1/{city}",
            timeout=2  # 2秒超时
        )
        response.raise_for_status()  # 非2xx状态码会抛出异常
        # 成功：返回真实数据
        data = response.json()
        print(f"[正常] 获取到天气数据: {data}")
        return data
    except requests.exceptions.Timeout:
        print(f"[降级] 天气服务超时，返回缓存数据")
        return CACHE_DATA
    except requests.exceptions.ConnectionError:
        print(f"[降级] 网络连接失败，返回缓存数据")
        return CACHE_DATA
    except requests.exceptions.RequestException as e:
        print(f"[降级] 请求异常: {e}，返回缓存数据")
        return CACHE_DATA
# 模拟连续调用
for i in range(5):
    result = fetch_weather("beijing")
    print(f"第{i+1}次结果: {result}\n")
    time.sleep(1)

关键点：使用 timeout 防止无限等待，用 try/except 捕获异常后执行降级逻辑（返回缓存、默认值或空数据）。

API版本降级（版本回退）

假设你调用了 v2 的API，但如果 v2 不可用，自动回退到 v1。

示例脚本（Python + 版本检测与回退）

import requests
def api_call_v2(params):
    """尝试调用v2版本API"""
    url = "https://api.example.com/v2/endpoint"
    response = requests.get(url, params=params, timeout=3)
    response.raise_for_status()
    return response.json()
def api_call_v1(params):
    """v1版本API（备选）"""
    url = "https://api.example.com/v1/endpoint"
    response = requests.get(url, params=params, timeout=5)
    response.raise_for_status()
    return response.json()
def get_data_with_fallback(params):
    """
    优先v2，v2失败则降级到v1
    """
    try:
        return api_call_v2(params)
    except Exception as e:
        print(f"v2请求失败: {e}")
        print("正在降级到 v1...")
        try:
            return api_call_v1(params)
        except Exception as e2:
            print(f"v1也失败了: {e2}")
            return None  # 或抛出异常
# 使用示例
params = {"user_id": 123}
result = get_data_with_fallback(params)
if result:
    print("成功获取数据:", result)
else:
    print("所有版本均失败，请检查服务")

关键点：定义两个独立的调用函数，优先调用高版本，捕获异常后调用低版本，注意低版本API的签名和返回值可能不同，需要做适配。

更高级的降级策略（生产环境常用）

1. 渐进式超时：先快（短超时）后慢（长超时），防止拖垮系统。
2. 降级计数器：连续失败N次后，直接降级（避免频繁尝试）。
3. 异步降级：在失败时记录日志、发送告警。
4. 动态配置：通过配置文件或环境变量控制降级开关（如 ENABLE_FALLBACK=true）。

示例：带有降级计数器和动态配置的脚本（伪代码）

import requests
import time
# 全局降级状态
class Degrader:
    def __init__(self, threshold=3, cooldown=10):
        self.fail_count = 0
        self.threshold = threshold
        self.cooldown = cooldown
        self.last_fail_time = 0
    def should_degrade(self):
        """是否触发降级"""
        now = time.time()
        # 还在冷却期内，继续降级
        if now - self.last_fail_time < self.cooldown:
            return True
        if self.fail_count >= self.threshold:
            self.last_fail_time = now
            return True
        return False
    def record_success(self):
        self.fail_count = 0  # 成功则重置
    def record_failure(self):
        self.fail_count += 1
# 使用
degrader = Degrader(threshold=5, cooldown=30)
def safe_call_api():
    if degrader.should_degrade():
        print("系统已降级，返回默认值")
        return {"default": True}
    try:
        result = requests.get("https://api.example.com/data", timeout=2)
        degrader.record_success()
        return result.json()
    except:
        degrader.record_failure()
        return {"default": True}

最后一点重要提示：降级不是把错误吞掉就完事了。必须记录日志（如 warn 级别）并设置监控告警，否则你可能永远不知道服务已经处于降级状态，导致数据不准或功能异常。

如果你说的是降级某个服务的API访问权限（比如关闭某个付费API），那是“关闭”或“禁用”API，而不是“降级”，如有这种情况，请补充说明。