📢 专栏导语

欢迎来到《企业级Python自动化运维实战》专栏的第四篇！在前几篇文章中，我们从一个简单的单机脚本，一步步将其升级为支持多服务器并发巡检、SSH密钥认证、结构化数据解析的企业级工具。然而，一个真正的SRE（站点可靠性工程）平台，仅有“实时巡检”是远远不够的

---

🎯 本篇目标：为巡检系统装上“记忆”

在前面的版本中，我们已经完成了从 SSH 采集、指标解析、异常分析到 SQLite 存储和 HTML 报告的完整链路：

服务器 → SSH采集 → 指标解析 → 异常分析 → SQLite保存 → HTML报告

但有一个明显的瓶颈：所有数据只能通过登录服务器运行脚本查看，外部系统无法访问。

在企业内部，SRE 平台通常不允许运维人员直接登录服务器执行命令。真实的交互方式是：

浏览器 / 其他系统 → API 接口 → 查询监控数据

因此，版本的目标是：将巡检系统从一个"可执行脚本"升级为一个"后台服务"。

                 用户
                  |
             HTTP请求
                  |
              FastAPI
                  |
        -------------------
        |                 |
    查询接口          手动触发巡检
        |                 |
    SQLite数据库       main.py

升级完成后，对外暴露以下接口：

接口路径	方法	功能
/api/history	GET	查询所有服务器最近巡检记录
/api/history/{server_name}	GET	查询指定服务器的巡检记录
/api/check	POST	手动触发一次巡检
/api/health	GET	健康检查（供 K8s / LB 探测）

---

🔧 一、安装 FastAPI 环境

进入项目虚拟环境，安装依赖：

pip install fastapi uvicorn

💡 知识点补充

组件	作用
FastAPI	现代 Python Web 框架，基于 Python 类型提示，自动生成 OpenAPI 文档，性能接近 Go/Node.js
Uvicorn	ASGI 服务器，负责监听端口、接收 HTTP 请求并转发给 FastAPI 应用

ASGI（Asynchronous Server Gateway Interface）是 Python 异步 Web 服务器网关接口标准，是 WSGI 的异步升级版。FastAPI 基于 ASGI 运行，天然支持异步处理。

FastAPI 的核心优势：

• 🚀 高性能：基于 Starlette（HTTP）和 Pydantic（数据校验），性能媲美 Node.js 和 Go
• 📖 自动文档：内置 Swagger UI（/docs）和 ReDoc（/redoc）
• ✅ 类型校验：利用 Python Type Hints 自动做参数校验和序列化
• 🔌 标准兼容：完全兼容 OpenAPI 和 JSON Schema

---

📁 二、调整项目结构

server-inspection
├── main.py              # 原巡检入口
├── database/            # 数据库模型与连接
├── collectors/          # SSH采集模块
├── analyzer/            # 异常分析模块
├── report/              # 报告生成模块
├── api/                 # 🆕 新增：API接口层
│   ├── main.py          # FastAPI 应用入口
│   ├── history.py       # 历史查询接口
│   ├── inspection.py    # 手动触发巡检接口
│   └── health.py        # 健康检查接口

💡 知识点补充：为什么要分层？

在大型项目中，将所有接口写在一个文件会导致：

• 代码臃肿，难以维护
• 多人协作时冲突频繁
• 职责不清晰

FastAPI 提供了 APIRouter 机制，允许我们将路由拆分到不同模块，最终通过 include_router() 注册到主应用。这类似于 Flask 中的 Blueprint。

---

🚀 三、创建 FastAPI 入口

api/main.py

from fastapi import FastAPI

app = FastAPI(
    title="Server Inspection API",
    description="SRE巡检平台接口",
    version="1.0"
)

@app.get("/")
def index():
    return {
        "message": "Server Inspection API Running"
    }

💡 知识点补充

参数	作用
title	显示在 Swagger 文档标题
description	显示在文档描述区域
version	API 版本号，便于多版本管理

@app.get("/") 是一个装饰器，它将下方的函数绑定到 HTTP GET 请求的 / 路径。FastAPI 支持的路由装饰器包括：

• @app.get() — 查询数据
• @app.post() — 创建/提交数据
• @app.put() — 更新数据
• @app.delete() — 删除数据
• @app.patch() — 部分更新

---

▶️ 四、启动 API 服务

在项目根目录执行：

uvicorn api.main:app --reload

命令解析

部分	含义
uvicorn	ASGI 服务器
api.main	Python 模块路径（api 目录下的 main.py）
:app	模块中的 FastAPI 实例变量名
--reload	开发模式，文件变更自动重启（生产环境不要使用）

看到如下输出即表示启动成功：

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx]
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

浏览器打开 http://127.0.0.1:8000，看到：

{
  "message": "Server Inspection API Running"
}

---

📖 五、使用 Swagger 测试接口

FastAPI 最让人兴奋的特性之一：自动生成可交互的 API 文档。

打开浏览器访问：

http://127.0.0.1:8000/docs

你会看到 Swagger UI 界面，列出了所有已注册的接口。

使用步骤：

1. 点击接口右侧的 Try it out 按钮
2. 填写参数（如果有）
3. 点击 Execute
4. 查看 Response

💡 知识点补充

FastAPI 提供两套自动文档：

路径	文档风格	适用场景
/docs	Swagger UI（交互式）	开发调试、接口测试
/redoc	ReDoc（阅读式）	对外发布 API 文档

这些文档基于 OpenAPI 3.0 规范自动生成，无需手动编写 YAML/JSON。

---

📊 六、开发历史查询接口

历史数据存储在 SQLite 中，查询流程为：

HTTP请求 → APIRouter → 查询数据库 → ORM对象 → 转为字典 → FastAPI序列化为JSON → 返回

api/history.py

from database.db import get_db
from fastapi import APIRouter
from database.models import InspectionRecord

router = APIRouter()

@router.get("/history")
def history():
    db = get_db()
    records = db.query(InspectionRecord).all()
    result=[]
    for item in records:
        result.append({
            "server":item.server_name,
            "cpu":item.cpu,
            "memory":item.memory,
            "disk":item.disk,
            "status":item.status,
            "time":str(item.create_time)
        })
    db.close()
    return result

@router.get("/history/{server_name}")
def server_history(server_name:str):
    db = get_db()
    records = db.query(InspectionRecord).filter(InspectionRecord.server_name==server_name).all()
    result=[]
    for item in records:
        result.append({
            "server": item.server_name,
            "cpu": item.cpu,
            "memory": item.memory,
            "disk": item.disk,
            "status": item.status,
            "time": str(item.create_time)
        })
    db.close()
    return result

💡 知识点补充

1. APIRouter 是什么？

APIRouter 是 FastAPI 的路由分组工具。它和 FastAPI() 实例拥有几乎相同的方法（.get()、.post() 等），但它不能独立运行，必须被注册到主应用。

router = APIRouter()  # 创建子路由
app.include_router(router)  # 注册到主应用

2. 路径参数

@router.get("/history/{server_name}")
def server_history(server_name: str):

{server_name} 是路径参数，FastAPI 会自动从 URL 中提取值并传入函数。类型标注 :str 告诉 FastAPI 做类型校验——如果传入的是数字，会自动尝试转换或返回 422 错误。

3. ORM 对象 → 字典 → JSON

SQLAlchemy 查询返回的是 ORM 对象，不能直接 JSON 序列化。我们手动将其转为字典，FastAPI 再将字典转为 JSON 响应。

在更复杂的场景中，可以使用 Pydantic 的 response_model 来自动完成序列化。

---

🔗 七、注册路由

创建了接口文件后，还需要将其"挂载"到主应用上：

api/main.py（补充）

from api.history import router as history_router
app.include_router(history_router, prefix="/api")

参数	作用
prefix="/api"	给所有路由添加统一前缀，避免路径冲突
tags=["历史"]	在 Swagger 文档中分组显示（可选）

重启服务后访问：

http://127.0.0.1:8000/api/history
http://127.0.0.1:8000/api/history/master

---

filter vs filter_by 对比

在查询指定服务器时，SQLAlchemy 提供了两种过滤方式：

对比项	filter_by	filter
语法	filter_by(字段名=值)	filter(模型.字段名 == 值)
支持的运算符	只支持 ==（等于）	支持 ==、!=、>、<、>=、<=、like、in_ 等
多条件	filter_by(a=1, b=2)	filter(模型.a==1, 模型.b==2) 或用 .and_()
关联表查询	❌ 不支持	✅ 支持（可跨表关联）
灵活度	低，适合简单等值查询	高，适合复杂条件查询

实际开发建议：简单等值查询用 filter_by 更简洁，复杂条件查询用 filter 更灵活。

---

⚡ 九、增加手动触发巡检接口

巡检平台默认是定时执行，但运维人员经常需要"立即看到当前状态"。因此我们添加一个手动触发接口。

api/inspection.py

from fastapi import APIRouter
import subprocess
import os

path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
router=APIRouter()

@router.post("/check")
def start_check():
    subprocess.Popen(["python",os.path.join(path,"main.py")])
    return {"message":"inspection started"}

💡 知识点补充

1. 为什么用 subprocess.Popen 而不是直接 import main？

方式	问题
import main; main.start()	FastAPI 和巡检在同一进程，巡检耗时 10 分钟，接口就会卡住 10 分钟
subprocess.Popen(...)	启动独立进程，接口立即返回，巡检在后台运行

2. subprocess.run() vs subprocess.Popen() 对比

对比项	run()	Popen()
是否阻塞	✅ 阻塞，等命令跑完才返回	❌ 非阻塞，立即返回
返回值	CompletedProcess 对象	Popen 对象
输出获取	自动收集到 result.stdout	需手动 process.stdout.read()
实时读取	❌ 不支持	✅ 支持逐行读取
发送输入	通过 input= 参数一次性传入	通过 process.stdin.write() 随时写入
提前终止	❌ 不支持	✅ process.kill() / process.terminate()
检查运行状态	❌ 不支持（已经跑完了）	✅ process.poll() 检查是否还在跑
代码复杂度	低，一行搞定	高，需要手动管理

3. 为什么用 POST 而不是 GET？

根据 HTTP 语义：

• GET：幂等，只读取数据，不产生副作用
• POST：非幂等，会触发服务端操作（如创建任务、触发巡检）

手动触发巡检是一个"有副作用"的操作，所以用 POST。

注册到主程序

api/main.py（补充）

from api.inspection import router as inspection_router
app.include_router(inspection_router, prefix="/api", tags=["巡检"])

测试

在 Swagger 中：

POST /api/check

返回：

{
  "message": "inspection started"
}

后台巡检开始执行，稍后可通过 /api/history 查看新记录验证。

---

💓 十、增加健康检查接口

在企业级服务中，健康检查接口是必需品：

• Kubernetes：通过 livenessProbe 和 readinessProbe 探测容器是否存活
• 负载均衡器（Nginx、ALB）：通过健康检查判断后端是否可用
• 监控系统：定期拨测服务状态

api/health.py

from fastapi import APIRouter
router = APIRouter()

@router.get("/health")
def health():
    return {"message":"ok"}

注册：

from api.health import router as health_router
app.include_router(health_router, tags=["系统"])

💡 知识点补充：K8s 探针类型

探针类型	作用	失败后果
livenessProbe	判断容器是否存活	重启容器
readinessProbe	判断容器是否就绪（能接收流量）	从 Service 摘除
startupProbe	判断应用是否启动完成	启动期间不触发前两个探针

K8s 配置示例：

livenessProbe:
  httpGet:
    path: /api/health
    port: 8000
  initialDelaySeconds: 5
  periodSeconds: 10

---

📦 十一、完整的 api/main.py

将所有路由注册完毕后，最终的入口文件如下：

from fastapi import FastAPI
from api.history import router as history_router
from api.inspection import router as inspection_router
from api.health import router as health_router

app = FastAPI(
    title="Server Inspection API",
    description="SRE巡检平台接口",
    version="2.3"
)

@app.get("/")
def index():
    return {"message": "Server Inspection API Running"}

app.include_router(history_router, prefix="/api", tags=["历史查询"])
app.include_router(inspection_router, prefix="/api", tags=["巡检"])
app.include_router(health_router, prefix="/api",tags=["系统"])

---

✅ 十二、验证清单

接口	方法	预期结果
GET /	GET	返回 API 运行状态
GET /api/history	GET	返回所有巡检记录 JSON
GET /api/history/master	GET	返回 master 服务器的巡检记录
POST /api/check	POST	触发后台巡检，立即返回
GET /api/health	GET	返回 {"message":"ok"}
GET /docs	GET	Swagger 交互文档
GET /redoc	GET	ReDoc 文档

---

🧠 本节核心知识总结

知识点	要点
FastAPI	基于 ASGI 的高性能 Python Web 框架，自动文档 + 类型校验
Uvicorn	ASGI 服务器，负责监听端口和处理 HTTP 连接
APIRouter	路由分组工具，实现接口模块化
路径参数	{param} 语法，自动提取 URL 中的值
filter vs filter_by	filter 更灵活，filter_by 更简洁
subprocess.Popen	异步启动子进程，避免阻塞主服务
健康检查	生产必备，供 K8s/LB 探测服务状态

---

🔮 下一阶段预告

Vue 前端管理页面

目前我们只能通过 Swagger 或 curl 访问接口，对于非技术人员并不友好。下一阶段我们将使用 Vue.js 构建一个可视化管理前端，实现：

• 📊 巡检数据仪表盘
• 📋 历史查询与筛选
• 🚀 一键触发巡检
• 📈 趋势图表展示

从"能用"到"好用"，继续进化 🚀