HTTP 接口（requests / httpx）

对接监控、CMDB、K8s API、内部运维平台，几乎都走 HTTP/HTTPS。推荐 httpx（支持同步/异步、HTTP/2）；生态更广的 requests 同样够用。项目内用 uv 管理依赖。

---

安装依赖

cd ~/python-for-ops
uv add httpx
# 或: uv add requests

---

GET 与超时

#!/usr/bin/env python3
"""HTTP 探活与 JSON API 调用示例（httpx）。"""
import httpx
import sys

def health_check(url: str, timeout: float = 5.0) -> bool:
    try:
        r = httpx.get(url, timeout=timeout, follow_redirects=True)
        return r.status_code == 200
    except httpx.RequestError as e:
        print(f"请求失败: {e}", file=sys.stderr)
        return False

def get_json(url: str, headers: dict | None = None) -> dict:
    headers = headers or {}
    r = httpx.get(url, headers=headers, timeout=30.0)
    r.raise_for_status()  # 4xx/5xx 抛 httpx.HTTPStatusError
    return r.json()

---

带 Token 与 TLS

import os
import httpx

API_BASE = os.environ.get("OPS_API_BASE", "https://ops.example.com/api")
TOKEN = os.environ["OPS_API_TOKEN"]  # 勿写死在代码里

def ops_client() -> httpx.Client:
    return httpx.Client(
        base_url=API_BASE,
        headers={"Authorization": f"Bearer {TOKEN}"},
        timeout=httpx.Timeout(10.0, connect=5.0),
        verify=True,  # 内网自签证书可指定 verify="/path/ca.pem"
    )

def list_hosts(client: httpx.Client) -> list[dict]:
    r = client.get("/v1/hosts", params={"page": 1, "size": 100})
    r.raise_for_status()
    return r.json().get("items", [])

with ops_client() as client:
    hosts = list_hosts(client)
    print(f"共 {len(hosts)} 台主机")

---

写操作、重试与幂等键

import random
import time
import httpx


RETRYABLE_STATUS = {429, 500, 502, 503, 504}


def post_with_retry(
    client: httpx.Client,
    url: str,
    json_body: dict,
    idempotency_key: str,
    attempts: int = 3,
) -> httpx.Response:
    """只对服务端支持幂等键的变更接口重试。

    网络中断或 5xx 不代表服务端没有完成操作；服务端必须按
    Idempotency-Key 保存并复用第一次的处理结果，客户端才能安全重试。
    """
    last_error: Exception | None = None
    last_response: httpx.Response | None = None
    for attempt in range(1, attempts + 1):
        try:
            r = client.post(
                url,
                json=json_body,
                headers={"Idempotency-Key": idempotency_key},
            )
            if r.status_code not in RETRYABLE_STATUS:
                r.raise_for_status()  # 4xx 等不可恢复错误立即返回给调用方
                return r
            last_response = r
        except httpx.RequestError as e:
            last_error = e

        if attempt == attempts:
            break

        # 优先尊重服务端限流窗口；无该头时使用指数退避加随机抖动。
        retry_after = (last_response.headers.get("Retry-After") if last_response else None)
        try:
            delay = float(retry_after) if retry_after else min(30, 2 ** (attempt - 1))
        except ValueError:
            delay = min(30, 2 ** (attempt - 1))
        time.sleep(delay + random.uniform(0, 0.5))

    if last_response is not None:
        last_response.raise_for_status()
    raise RuntimeError(f"请求在 {attempts} 次尝试后仍失败: {last_error}")


# 用共享 Client 复用连接；变更请求的幂等键应来自调用任务 ID，而不是随机重试 ID。
with httpx.Client(timeout=httpx.Timeout(15.0, connect=5.0)) as client:
    response = post_with_retry(
        client,
        "https://ops.example.com/v1/restarts",
        {"host": "web-01", "service": "nginx"},
        idempotency_key="change-2026-07-23-0001",
    )
    print(response.json())

---

requests 等价写法

import requests

session = requests.Session()
session.headers.update({"Authorization": f"Bearer {TOKEN}"})
r = session.get(f"{API_BASE}/v1/hosts", timeout=(5, 30))  # connect, read
r.raise_for_status()
data = r.json()

---

K8s API 只读示例

import httpx
from pathlib import Path

KUBE_TOKEN = Path("/var/run/secrets/kubernetes.io/serviceaccount/token").read_text()
KUBE_CA = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
NS = "default"

r = httpx.get(
    f"https://kubernetes.default.svc/api/v1/namespaces/{NS}/pods",
    headers={"Authorization": f"Bearer {KUBE_TOKEN}"},
    verify=KUBE_CA,
    timeout=10.0,
)
r.raise_for_status()
print(f"Pod 数量: {len(r.json()['items'])}")

项	建议
超时	始终设置 connect + read
Session	多请求复用连接
密钥	环境变量或 K8s SA
日志	勿打印完整 Token/响应里的密码
变更重试	仅在服务端支持幂等键时自动重试

提示

第四阶段后续章节：SSH 跑远程命令；通用 API 模式（分页、限流、重试）见 04-api-automation.md。