第三方 API 自动化通用模式

对接 GitLab、Harbor、云厂商、内部 CMDB 时，模式高度相似：认证（Token）→ 分页拉取 → 限流与重试 → 统一错误处理。封装一个小客户端，比每处 copy 请求代码更易维护。

---

客户端骨架

#!/usr/bin/env python3
"""可复用的 API Client：Token、分页、重试。"""
from __future__ import annotations

import os
import time
from typing import Any, Iterator

import httpx

class OpsApiClient:
    def __init__(
        self,
        base_url: str | None = None,
        token: str | None = None,
        timeout: float = 30.0,
        max_retries: int = 3,
    ):
        self.base_url = (base_url or os.environ["API_BASE_URL"]).rstrip("/")
        self.token = token or os.environ["API_TOKEN"]
        self.max_retries = max_retries
        self._client = httpx.Client(
            base_url=self.base_url,
            headers={
                "Authorization": f"Bearer {self.token}",
                "Accept": "application/json",
            },
            timeout=timeout,
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> OpsApiClient:
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

    def request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
        last_exc: Exception | None = None
        for attempt in range(self.max_retries):
            try:
                r = self._client.request(method, path, **kwargs)
                if r.status_code == 429:
                    retry_after = int(r.headers.get("Retry-After", "2"))
                    time.sleep(retry_after)
                    continue
                if r.status_code >= 500:
                    time.sleep(2 ** attempt)
                    continue
                r.raise_for_status()
                return r
            except (httpx.RequestError, httpx.HTTPStatusError) as e:
                last_exc = e
                time.sleep(2 ** attempt)
        raise RuntimeError(f"API 请求失败 {method} {path}: {last_exc}")

    def get_json(self, path: str, **kwargs: Any) -> Any:
        return self.request("GET", path, **kwargs).json()

---

分页迭代

class OpsApiClient:
    # ... 上文方法 ...

    def paginate(
        self,
        path: str,
        *,
        page_param: str = "page",
        size_param: str = "per_page",
        page_size: int = 100,
        items_key: str = "items",
    ) -> Iterator[dict]:
        """假设响应: {"items": [...], "total": N}；按平台调整字段名。"""
        page = 1
        while True:
            data = self.get_json(
                path,
                params={page_param: page, size_param: page_size},
            )
            items = data.get(items_key, [])
            if not items:
                break
            yield from items
            if len(items) < page_size:
                break
            page += 1

# 使用：拉取全部项目
with OpsApiClient() as api:
    for project in api.paginate("/v4/projects", items_key="items"):
        print(project.get("id"), project.get("name"))

---

同步写入（幂等）

def ensure_label(api: OpsApiClient, project_id: int, label_name: str) -> None:
    """存在则跳过，不存在则创建 — 脚本可重复执行。"""
    labels = api.get_json(f"/v4/projects/{project_id}/labels")
    if any(lb["name"] == label_name for lb in labels):
        return
    api.request(
        "POST",
        f"/v4/projects/{project_id}/labels",
        json={"name": label_name, "color": "#6699cc"},
    )

---

实践要点

模式	要点
Token	环境变量；CI 用 Secret
分页	yield from 生成器，省内存
429/5xx	退避 + Retry-After
幂等	GET 查询再 POST；或用平台 upsert API
日志	捕获 RuntimeError，勿打印 Token

第四阶段覆盖 Socket 探活 → HTTP → SSH → API 客户端。第五阶段对接 GitLab、Harbor、Docker、K8s 等平台。

提示

将 OpsApiClient 放入 lib/api.py；Token 与 base_url 从环境变量读取（勿提交 Git）。