—— 35课知识集大成之作
"""自动化运维平台架构：
┌──────────────────────────────────────────┐
│ FastAPI Web 仪表盘 │
│ ┌──────┐ ┌──────┐ ┌──────┐ │
│ │ 监控 │ │ 告警 │ │ 日志 │ │
│ └──┬───┘ └──┬───┘ └──┬───┘ │
│ │ │ │ │
│ ┌──▼────────▼────────▼───┐ │
│ │ SQLite 数据层 │ │
│ └──┬────────┬────────┬───┘ │
│ │ │ │ │
│ ┌──▼──┐ ┌───▼──┐ ┌───▼──┐ │
│ │psutil│ │邮件 │ │配置 │ │
│ │监控 │ │通知 │ │管理 │ │
│ └─────┘ └──────┘ └──────┘ │
└──────────────────────────────────────────┘
技术栈：
- 系统监控：psutil（第24课）
- Web框架：FastAPI（第26课）
- 数据库：SQLite（第22课）/ SQLAlchemy（第30课）
- 配置管理：dataclass + 环境变量（第32课）
- 定时任务：schedule（第19课）
- 日志：logging（第12课）
- 并发：threading（第28课）
"""import sqlite3
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional, List
import json
@dataclass
class MetricRecord:
"""监控指标记录"""
id: Optional[int] = None
timestamp: str = ""
cpu_percent: float = 0.0
memory_percent: float = 0.0
disk_percent: float = 0.0
network_sent: int = 0
network_recv: int = 0
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
@dataclass
class AlertRecord:
"""告警记录"""
id: Optional[int] = None
timestamp: str = ""
level: str = "warning" # warning / critical
metric: str = ""
value: float = 0.0
threshold: float = 0.0
message: str = ""
notified: bool = False
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
class OpsDatabase:
"""运维平台数据库"""
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_tables()
def _init_tables(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
cpu_percent REAL,
memory_percent REAL,
disk_percent REAL,
network_sent INTEGER,
network_recv INTEGER
);
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
level TEXT NOT NULL,
metric TEXT NOT NULL,
value REAL,
threshold REAL,
message TEXT,
notified INTEGER DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_metrics_ts ON metrics(timestamp);
CREATE INDEX IF NOT EXISTS idx_alerts_ts ON alerts(timestamp);
""")
self.conn.commit()
def insert_metric(self, metric: MetricRecord):
self.conn.execute(
"""INSERT INTO metrics (timestamp, cpu_percent, memory_percent,
disk_percent, network_sent, network_recv)
VALUES (?, ?, ?, ?, ?, ?)""",
(metric.timestamp, metric.cpu_percent, metric.memory_percent,
metric.disk_percent, metric.network_sent, metric.network_recv)
)
self.conn.commit()
def insert_alert(self, alert: AlertRecord):
self.conn.execute(
"""INSERT INTO alerts (timestamp, level, metric, value, threshold, message, notified)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(alert.timestamp, alert.level, alert.metric, alert.value,
alert.threshold, alert.message, 1 if alert.notified else 0)
)
self.conn.commit()
def get_recent_metrics(self, minutes=60):
rows = self.conn.execute(
"""SELECT * FROM metrics
WHERE timestamp >= datetime('now', ?)
ORDER BY timestamp DESC""",
(f"-{minutes} minutes",)
).fetchall()
return [dict(r) for r in rows]
def get_recent_alerts(self, limit=20):
rows = self.conn.execute(
"SELECT * FROM alerts ORDER BY timestamp DESC LIMIT ?", (limit,)
).fetchall()
return [dict(r) for r in rows]import psutil
import time
import threading
from typing import Callable, Optional
@dataclass
class AlertRule:
"""告警规则"""
metric: str # cpu / memory / disk
threshold: float # 阈值
level: str # warning / critical
compare: str = "gt" # gt(大于) / lt(小于)
class MonitorEngine:
"""监控引擎"""
def __init__(self, db: OpsDatabase, interval=60):
self.db = db
self.interval = interval
self.rules: List[AlertRule] = []
self._running = False
self._thread: Optional[threading.Thread] = None
self.on_alert: Optional[Callable] = None
def add_rule(self, rule: AlertRule):
self.rules.append(rule)
def collect(self) -> MetricRecord:
"""采集系统指标"""
cpu = psutil.cpu_percent(interval=1)
mem = psutil.virtual_memory()
disk = psutil.disk_usage("/")
net = psutil.net_io_counters()
return MetricRecord(
cpu_percent=cpu,
memory_percent=mem.percent,
disk_percent=disk.percent,
network_sent=net.bytes_sent,
network_recv=net.bytes_recv,
)
def check_alerts(self, metric: MetricRecord):
"""检查告警规则"""
values = {
"cpu": metric.cpu_percent,
"memory": metric.memory_percent,
"disk": metric.disk_percent,
}
for rule in self.rules:
value = values.get(rule.metric, 0)
triggered = (
value > rule.threshold if rule.compare == "gt"
else value < rule.threshold
)
if triggered:
alert = AlertRecord(
level=rule.level,
metric=rule.metric,
value=value,
threshold=rule.threshold,
message=f"{rule.metric} = {value:.1f}% (阈值: {rule.threshold}%)",
)
self.db.insert_alert(alert)
if self.on_alert:
self.on_alert(alert)
def start(self):
"""启动监控"""
self._running = True
self._thread = threading.Thread(target=self._run_loop, daemon=True)
self._thread.start()
print(f"✅ 监控引擎已启动 (间隔: {self.interval}s)")
def stop(self):
self._running = False
if self._thread:
self._thread.join(timeout=5)
print("✅ 监控引擎已停止")
def _run_loop(self):
while self._running:
metric = self.collect()
self.db.insert_metric(metric)
self.check_alerts(metric)
time.sleep(self.interval)from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI(title="运维平台 API", version="1.0.0")
# 全局数据库（实际项目用依赖注入）
db = OpsDatabase()
@app.get("/api/metrics")
def get_metrics(minutes: int = 60):
"""获取最近监控数据"""
return {"data": db.get_recent_metrics(minutes)}
@app.get("/api/alerts")
def get_alerts(limit: int = 20):
"""获取最近告警"""
return {"data": db.get_recent_alerts(limit)}
@app.get("/api/status")
def get_status():
"""获取当前系统状态"""
cpu = psutil.cpu_percent(interval=0.5)
mem = psutil.virtual_memory()
disk = psutil.disk_usage("/")
return {
"cpu_percent": cpu,
"memory_percent": mem.percent,
"memory_used_gb": round(mem.used / 1024**3, 1),
"memory_total_gb": round(mem.total / 1024**3, 1),
"disk_percent": disk.percent,
"disk_used_gb": round(disk.used / 1024**3, 1),
"disk_total_gb": round(disk.total / 1024**3, 1),
}
@app.post("/api/collect")
def manual_collect():
"""手动触发采集"""
engine = MonitorEngine(db)
metric = engine.collect()
db.insert_metric(metric)
return {"message": "采集成功", "data": asdict(metric)}#!/usr/bin/env python3
"""第35课 毕业项目验证"""
import psutil
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional
@dataclass
class MetricRecord:
id: Optional[int] = None
timestamp: str = ""
cpu_percent: float = 0.0
memory_percent: float = 0.0
disk_percent: float = 0.0
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
@dataclass
class AlertRecord:
id: Optional[int] = None
timestamp: str = ""
level: str = "warning"
metric: str = ""
value: float = 0.0
threshold: float = 0.0
message: str = ""
notified: bool = False
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
import sqlite3
class OpsDatabase:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
cpu_percent REAL,
memory_percent REAL,
disk_percent REAL
);
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
level TEXT NOT NULL,
metric TEXT NOT NULL,
value REAL,
threshold REAL,
message TEXT,
notified INTEGER DEFAULT 0
);
""")
self.conn.commit()
def insert_metric(self, m):
self.conn.execute(
"INSERT INTO metrics (timestamp, cpu_percent, memory_percent, disk_percent) VALUES (?,?,?,?)",
(m.timestamp, m.cpu_percent, m.memory_percent, m.disk_percent))
self.conn.commit()
def insert_alert(self, a):
self.conn.execute(
"INSERT INTO alerts (timestamp, level, metric, value, threshold, message, notified) VALUES (?,?,?,?,?,?,?)",
(a.timestamp, a.level, a.metric, a.value, a.threshold, a.message, 1 if a.notified else 0))
self.conn.commit()
def get_recent_metrics(self, limit=10):
return [dict(r) for r in self.conn.execute(
"SELECT * FROM metrics ORDER BY id DESC LIMIT ?", (limit,)).fetchall()]
def get_recent_alerts(self, limit=10):
return [dict(r) for r in self.conn.execute(
"SELECT * FROM alerts ORDER BY id DESC LIMIT ?", (limit,)).fetchall()]
def test_data_collection():
db = OpsDatabase()
cpu = psutil.cpu_percent(interval=0.1)
mem = psutil.virtual_memory()
disk = psutil.disk_usage("/")
metric = MetricRecord(cpu_percent=cpu, memory_percent=mem.percent, disk_percent=disk.percent)
db.insert_metric(metric)
results = db.get_recent_metrics()
assert len(results) == 1
assert results[0]["cpu_percent"] == cpu
print(f"✅ 数据采集测试通过 (CPU:{cpu}% MEM:{mem.percent}% DISK:{disk.percent}%)")
def test_alert_system():
db = OpsDatabase()
alert = AlertRecord(level="critical", metric="cpu", value=95.0, threshold=80.0,
message="CPU使用率95%超过阈值80%")
db.insert_alert(alert)
alerts = db.get_recent_alerts()
assert len(alerts) == 1
assert alerts[0]["level"] == "critical"
print("✅ 告警系统测试通过")
def test_api_endpoints():
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
db = OpsDatabase()
@app.get("/api/status")
def status():
cpu = psutil.cpu_percent(interval=0.1)
mem = psutil.virtual_memory()
return {"cpu": cpu, "memory": mem.percent}
@app.get("/api/metrics")
def metrics():
return {"data": db.get_recent_metrics()}
client = TestClient(app)
resp = client.get("/api/status")
assert resp.status_code == 200
data = resp.json()
assert "cpu" in data
assert "memory" in data
resp = client.get("/api/metrics")
assert resp.status_code == 200
print("✅ API端点测试通过")
def test_full_integration():
"""完整集成测试"""
db = OpsDatabase()
# 模拟采集3次
for i in range(3):
cpu = psutil.cpu_percent(interval=0.1)
mem = psutil.virtual_memory()
metric = MetricRecord(cpu_percent=cpu, memory_percent=mem.percent, disk_percent=50.0)
db.insert_metric(metric)
if cpu > 50:
alert = AlertRecord(level="warning", metric="cpu", value=cpu, threshold=50.0,
message=f"CPU {cpu}% > 50%")
db.insert_alert(alert)
metrics = db.get_recent_metrics(limit=3)
assert len(metrics) == 3
alerts = db.get_recent_alerts()
print(f"✅ 集成测试通过 (3次采集, {len(alerts)}条告警)")
# 输出最终报告
print("\n" + "=" * 50)
print("🎓 Python自动化课程 - 毕业验证")
print("=" * 50)
print(f"✅ 系统监控: CPU/内存/磁盘数据采集正常")
print(f"✅ 数据存储: SQLite 读写正常")
print(f"✅ 告警系统: 阈值检测正常")
print(f"✅ Web API: FastAPI 端点响应正常")
print(f"✅ 集成测试: 全链路打通")
print("=" * 50)
print("🎉 恭喜！你已完成全部35课学习！")
print("=" * 50)
if __name__ == "__main__":
test_data_collection()
test_alert_system()
test_api_endpoints()
test_full_integration()