Python 运维基础脚本#

1
# 读取文件所有行并统计行数
2
with open("/etc/passwd", 'r') as f:
3
    lines = f.readlines()
4
    print(f"行数: {len(lines)}")
5

6
# 逐行读取（大文件推荐）
7
with open("/var/log/messages", 'r') as f:
8
    for line in f:
9
        if 'error' in line.lower():
10
            print(line.strip())
11

12
# 写入文件
13
with open("/tmp/output.txt", 'w') as f:
14
    f.write("第一行\n")
15
    f.write("第二行\n")
16

17
# 追加写入
18
with open("/tmp/log.txt", 'a') as f:
19
    f.write("新的日志行\n")

1
import os
2

3
# 列出目录内容
4
print(os.listdir('/etc/'))
5

6
# 递归遍历目录树
7
for root, dirs, files in os.walk('/etc/'):
8
    for f in files:
9
        print(os.path.join(root, f))
10

11
# 判断路径类型
12
os.path.exists('/etc/passwd')    # 是否存在
13
os.path.isfile('/etc/passwd')    # 是否为文件
14
os.path.isdir('/etc/')           # 是否为目录

1
import argparse
2

3
parser = argparse.ArgumentParser(description='示例工具')
4
sub = parser.add_subparsers(dest='command')
5

6
# create 子命令
7
p_create = sub.add_parser('create', help='创建资源')
8
p_create.add_argument('-n', '--name', required=True, help='资源名称')
9
p_create.add_argument('-m', '--memory', type=int, default=512, help='内存大小(MB)')
10

11
# delete 子命令
12
p_del = sub.add_parser('delete', help='删除资源')
13
p_del.add_argument('-n', '--name', required=True, help='资源名称')
14

15
# list 子命令
16
sub.add_parser('list', help='列出所有资源')
17

18
args = parser.parse_args()
19

20
if args.command == 'create':
21
    print(f"创建 {args.name}, 内存 {args.memory}MB")
22
elif args.command == 'delete':
23
    print(f"删除 {args.name}")
24
elif args.command == 'list':
25
    print("列出所有资源")

1
import json
2

3
# 字典转 JSON 字符串
4
data = {"name": "server01", "status": "active"}
5
json_str = json.dumps(data, indent=2)
6
print(json_str)
7

8
# JSON 字符串转字典
9
parsed = json.loads('{"name": "server01", "status": "active"}')
10
print(parsed['name'])
11

12
# 读取 JSON 文件
13
with open('config.json', 'r') as f:
14
    config = json.load(f)
15

16
# 写入 JSON 文件
17
with open('output.json', 'w') as f:
18
    json.dump(data, f, indent=2, ensure_ascii=False)

1
import requests, json
2

3
# GET 请求
4
resp = requests.get('http://192.168.1.10:8774/v2.1/flavors',
5
                     headers={'X-Auth-Token': 'xxx'})
6
print(resp.status_code)
7
print(resp.json())
8

9
# POST 请求
10
body = {"flavor": {"name": "test", "ram": 1024, "vcpus": 2, "disk": 20}}
11
resp = requests.post('http://192.168.1.10:8774/v2.1/flavors',
12
                      data=json.dumps(body),
13
                      headers={'X-Auth-Token': 'xxx', 'Content-Type': 'application/json'})
14

15
# PUT 请求
16
resp = requests.put('http://192.168.1.10:9292/v2/images/xxx/file',
17
                     data=open('image.qcow2', 'rb').read(),
18
                     headers={'X-Auth-Token': 'xxx', 'Content-Type': 'application/octet-stream'})
19

20
# DELETE 请求
21
resp = requests.delete('http://192.168.1.10:5000/v3/users/xxx',
22
                        headers={'X-Auth-Token': 'xxx'})
23
print(resp.status_code)  # 204 表示成功

1
import yaml
2

3
# 读取 YAML 文件
4
with open('deployment.yaml', 'r', encoding='utf-8') as f:
5
    data = yaml.safe_load(f)
6
print(data['kind'])  # Deployment
7

8
# YAML 字符串转字典
9
yaml_str = """
10
apiVersion: v1
11
kind: Pod
12
metadata:
13
  name: nginx
14
"""
15
parsed = yaml.safe_load(yaml_str)
16

17
# 字典转 YAML 字符串
18
output = yaml.dump(parsed, default_flow_style=False)
19
print(output)

1
import subprocess
2

3
# 执行命令并获取输出
4
result = subprocess.run(['kubectl', 'get', 'pods'], capture_output=True, text=True)
5
print(result.stdout)
6
print(result.returncode)  # 0 表示成功
7

8
# 执行 shell 命令
9
result = subprocess.run('openstack server list', shell=True, capture_output=True, text=True)
10
print(result.stdout)
11

12
# 检查命令是否执行成功
13
if result.returncode != 0:
14
    print(f"命令失败: {result.stderr}")

1
import requests
2

3
# 基本异常捕获
4
try:
5
    resp = requests.get('http://192.168.1.10:5000/v3/users', timeout=5)
6
    resp.raise_for_status()  # 非 2xx 状态码抛异常
7
except requests.exceptions.ConnectionError:
8
    print("连接失败")
9
except requests.exceptions.Timeout:
10
    print("请求超时")
11
except requests.exceptions.HTTPError as e:
12
    print(f"HTTP 错误: {e}")
13

14
# K8S SDK 异常处理
15
from kubernetes.client.exceptions import ApiException
16

17
try:
18
    api.delete_namespaced_pod('nginx', 'default')
19
except ApiException as e:
20
    if e.status == 404:
21
        print("Pod 不存在")
22
    else:
23
        print(f"API 错误: {e.status} - {e.reason}")

1
import logging
2

3
logging.basicConfig(
4
    level=logging.INFO,
5
    format='%(asctime)s - %(levelname)s - %(message)s'
6
)
7
logger = logging.getLogger(__name__)
8

9
logger.info("服务启动")
10
logger.warning("配置文件缺失，使用默认值")
11
logger.error("连接数据库失败")

1
import os
2

3
# 读取环境变量（带默认值）
4
host = os.getenv('OS_AUTH_URL', 'http://192.168.1.10:5000/v3')
5
password = os.getenv('OS_PASSWORD', '000000')
6
debug = os.getenv('DEBUG', 'false').lower() == 'true'