import socket
from concurrent.futures import ThreadPoolExecutor
hosts = ["10.0.0.11", "10.0.0.12", "10.0.0.13"]
defcheck_port(host: str, port: int = 22) -> tuple[str, bool]:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
client.settimeout(2)
return host, client.connect_ex((host, port)) == 0with ThreadPoolExecutor(max_workers=10) as executor:
for host, ok in executor.map(check_port, hosts):
print(host, "open"if ok else"closed")
配置与数据处理
模块
安装
常见用途
PyYAML
uv add pyyaml
读取和生成 YAML 配置
Jinja2
uv add jinja2
渲染 Nginx、Prometheus、Kubernetes 模板
pydantic
uv add pydantic
校验配置、接口返回和脚本参数
python-dotenv
uv add python-dotenv
读取 .env 环境变量文件
openpyxl
uv add openpyxl
读写 Excel 巡检表、资产表
pandas
uv add pandas
汇总、清洗和分析表格数据
渲染配置模板
from jinja2 import Template
template = Template("""
server {
listen {{ port }};
server_name {{ domain }};
}
""")
print(template.render(port=80, domain="example.com"))
HTTP 与 API 调用
模块
安装
常见用途
requests
uv add requests
同步 HTTP 请求,简单稳定
httpx
uv add httpx
支持同步和异步请求
urllib.parse
标准库
URL 编码、解析和拼接
调用内部 API
import requests
response = requests.get(
"https://cmdb.example.com/api/hosts",
headers={"Authorization": "Bearer token"},
timeout=10,
)
response.raise_for_status()
for host in response.json()["items"]:
print(host["name"], host["ip"])
from kubernetes import client, config
config.load_kube_config()
api = client.CoreV1Api()
pods = api.list_namespaced_pod(namespace="default")
for pod in pods.items:
print(pod.metadata.name, pod.status.phase)
监控、日志与告警
模块
安装
常见用途
prometheus-client
uv add prometheus-client
暴露自定义 Exporter 指标
elasticsearch
uv add elasticsearch
查询 Elasticsearch 日志
redis
uv add redis
查询 Redis、写入缓存、实现简单锁
pymysql
uv add pymysql
查询 MySQL 运维数据
slack-sdk
uv add slack-sdk
发送 Slack 通知
暴露自定义指标
from prometheus_client import Gauge, start_http_server
import time
queue_depth = Gauge("demo_queue_depth", "Current queue depth")
start_http_server(9100)
whileTrue:
queue_depth.set(12)
time.sleep(15)
命令行工具与脚本体验
模块
安装
常见用途
typer
uv add typer
快速编写友好的命令行工具
click
uv add click
传统 CLI 框架,生态成熟
rich
uv add rich
彩色输出、表格、进度条、异常格式化
tqdm
uv add tqdm
循环进度条
APScheduler
uv add apscheduler
Python 内部定时任务
简单 CLI
import typer
app = typer.Typer()
@app.command()defcheck(host: str, port: int = 22):
print(f"check {host}:{port}")
if __name__ == "__main__":
app()