FastAPI 快速开发接口

前几阶段我们写脚本、调 API、批量巡检。第七阶段把能力封装成 HTTP 服务：CI/CD、告警系统、同事脚本都能通过接口调用，而不必 SSH 到某台机器手动跑脚本。

FastAPI 适合运维场景：类型提示 + 自动文档、性能好、与 Pydantic 集成校验请求体。

---

环境与最小示例

# 在练习项目中添加依赖
uv add fastapi uvicorn[standard]

# 开发模式启动（热重载）
uv run uvicorn main:app --reload --host 0.0.0.0 --port 8080

#!/usr/bin/env python3
"""main.py — 运维 API 最小骨架。"""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(
    title="Ops API",
    description="内部运维接口，勿暴露公网",
    version="0.1.0",
)

# ---------- 健康检查（K8s / LB 探活常用） ----------
@app.get("/health")
def health():
    return {"status": "ok"}

# ---------- 请求体模型：Pydantic 自动校验 ----------
class RestartServiceRequest(BaseModel):
    host: str = Field(..., description="目标主机 IP 或主机名")
    service: str = Field(..., min_length=1, description="systemd 单元名，如 nginx")
    reason: str = Field(default="", max_length=200)

@app.post("/ops/restart-service")
def restart_service(req: RestartServiceRequest):
    """
    示例接口：实际应调用 SSH / Ansible / 已有脚本。
    此处仅演示参数校验与响应结构。
    """
    # TODO: 接入第三阶段的 subprocess 或第四阶段的 SSH
    if not req.host.startswith(("10.", "192.168.")):
        raise HTTPException(status_code=400, detail="仅允许内网主机")
    return {
        "host": req.host,
        "service": req.service,
        "action": "restart",
        "message": f"已提交重启 {req.service}（示例，未真正执行）",
    }

启动后访问 http://127.0.0.1:8080/docs 可看到 Swagger UI，便于联调与交接。

---

查询类接口：封装已有函数

把前面章节的函数直接挂到路由上，保持业务逻辑与 HTTP 层分离：

from typing import Optional
from fastapi import Query

# 假设来自 03-system-automation 的封装
def kubectl_pods(namespace: str) -> list[dict]:
    import json, subprocess
    r = subprocess.run(
        ["kubectl", "get", "pods", "-n", namespace, "-o", "json"],
        capture_output=True, text=True, timeout=60,
    )
    if r.returncode != 0:
        raise RuntimeError(r.stderr)
    return json.loads(r.stdout).get("items", [])

@app.get("/k8s/pods")
def list_pods(
    namespace: str = Query(default="default", description="命名空间"),
    name_contains: Optional[str] = Query(default=None, description="Pod 名过滤"),
):
    try:
        items = kubectl_pods(namespace)
    except RuntimeError as e:
        raise HTTPException(status_code=502, detail=str(e))
    if name_contains:
        items = [p for p in items if name_contains in p["metadata"]["name"]]
    return {
        "namespace": namespace,
        "count": len(items),
        "pods": [p["metadata"]["name"] for p in items],
    }

---

统一错误与日志

import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger("ops-api")

@app.exception_handler(RuntimeError)
async def runtime_error_handler(request: Request, exc: RuntimeError):
    logger.exception("内部错误 path=%s", request.url.path)
    return JSONResponse(status_code=500, content={"detail": "内部执行失败，请查服务日志"})

# 启动前配置 logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

---

生产部署要点

项	建议
进程模型	uvicorn main:app --workers 2 --host 0.0.0.0 --port 8080
反向代理	Nginx / Ingress 做 TLS 与限流
配置	端口、KUBECONFIG 等走环境变量，不写死在代码
文档	内网可开 /docs；公网或敏感环境关闭 docs_url=None
健康检查	保持 /health 轻量，不依赖下游

# 生产可关闭公开文档
app = FastAPI(docs_url=None, redoc_url=None)

提示

下一篇为接口加上 API Key / Bearer 认证；再下一篇用 Webhook 接收 GitLab、Alertmanager 等事件，自动触发运维动作。