4、函数与模块:函数定义与参数、内置模块(os/sys/subprocess)、自定义模块与包
好,咱们今天聊聊函数和模块。说实话,这是Python自动化运维里最核心的东西。你想想看,没有函数,你的脚本就是一坨面条,改一个地方得翻半天。没有模块,你连个文件路径都拼不利索。
我个人习惯,写任何超过50行的脚本,第一件事就是拆函数。不是为了炫技,是为了后面少掉头发。
4.1 函数定义与参数
Python定义函数很简单,一个def搞定。但参数这块,坑不少。
4.1.1 基本定义
def check_disk_usage(mount_point):
"""检查磁盘使用率,返回百分比"""
import shutil
usage = shutil.disk_usage(mount_point)
return usage.used / usage.total * 100
print(f"根分区使用率: {check_disk_usage('/'):.1f}%")
嗯,这里要注意:函数一定要有文档字符串。我在项目里吃过亏,三个月后回来看自己的代码,完全想不起来这个函数是干嘛的。从那以后,每个函数我都写三行注释——干啥的、参数是啥、返回啥。
4.1.2 参数类型,别搞混了
| 参数类型 | 写法 | 我常用的场景 |
|---|---|---|
| 位置参数 | def func(a, b) |
必须传的参数,比如IP地址 |
| 默认参数 | def func(a, b=10) |
超时时间、重试次数 |
| 可变参数 | def func(*args) |
不确定数量的主机列表 |
| 关键字参数 | def func(**kwargs) |
传递配置字典 |
举个例子,我写过一个批量执行命令的函数:
def run_cmd_on_hosts(hosts, cmd, timeout=30, **extra_opts):
"""
在多台主机上执行命令
:param hosts: 主机列表,如 ['10.0.0.1', '10.0.0.2']
:param cmd: 要执行的命令
:param timeout: 超时时间,默认30秒
:param extra_opts: 额外参数,如 {'user': 'root', 'port': 2222}
"""
for host in hosts:
print(f"正在执行: {host}")
# 实际逻辑省略...
def func(lst=[]),多次调用会共享同一个列表。正确做法是用 None 然后在函数内部初始化。
4.2 内置模块:os / sys / subprocess
这三个模块,是运维脚本的「三驾马车」。说白了,你写的自动化脚本,90%都在跟这三样打交道:操作系统、解释器、外部命令。
4.2.1 os 模块——跟操作系统打交道
我每天都会用到的几个:
import os
# 路径操作——最常用
print(os.getcwd()) # 当前工作目录
print(os.listdir('/var/log')) # 列出目录
os.makedirs('/tmp/backup/2024', exist_ok=True) # 创建多级目录
# 环境变量——部署时必用
db_host = os.environ.get('DB_HOST', 'localhost')
db_port = int(os.environ.get('DB_PORT', '3306'))
# 文件信息
file_size = os.path.getsize('/etc/passwd')
file_mtime = os.path.getmtime('/etc/passwd')
你想想看,如果你的脚本要部署到不同环境,硬编码路径就是找死。用 os.environ 读取环境变量,配合默认值,这才是生产级的写法。
4.2.2 sys 模块——跟解释器打交道
sys 模块我主要用来做三件事:
import sys
# 1. 获取脚本参数
script_name = sys.argv[0]
action = sys.argv[1] if len(sys.argv) > 1 else 'help'
# 2. 控制退出码
if error_occurred:
sys.exit(1) # 非0表示异常退出
# 3. 添加自定义模块路径
sys.path.insert(0, '/opt/mycompany/lib')
sys.argv 配合 if-elif 就能实现简单的参数解析。当然,复杂场景还是用 argparse 模块更靠谱。
4.2.3 subprocess 模块——执行外部命令
这是运维脚本的重头戏。我建议你只用 subprocess.run(),其他变体基本可以忘掉。
import subprocess
# 执行命令,获取输出
result = subprocess.run(
['df', '-h', '/'],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
print("命令执行成功")
print(result.stdout)
else:
print(f"错误: {result.stderr}")
为什么会这样?因为 subprocess.run() 是 Python 3.5 之后推荐的统一接口。它返回一个 CompletedProcess 对象,包含了返回码、标准输出、标准错误,非常清晰。
shell=True 拼接用户输入!我曾经见过有人写 subprocess.run(f"ping {user_input}", shell=True),结果用户输入 ; rm -rf /……嗯,后果你懂的。
4.3 自定义模块与包
当你的函数越来越多,放在一个文件里就乱套了。这时候就需要拆模块、建包。
4.3.1 模块——一个 .py 文件就是一个模块
假设你写了一个 network_utils.py:
# network_utils.py
import socket
def check_port(host, port):
"""检查端口是否开放"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
result = sock.connect_ex((host, port))
sock.close()
return result == 0
def get_host_ip():
"""获取本机IP"""
return socket.gethostbyname(socket.gethostname())
在其他脚本里直接导入:
import network_utils
if network_utils.check_port('10.0.0.1', 22):
print("SSH端口开放")
4.3.2 包——一个目录就是一个包
当模块多了,就建个包。包就是一个包含 __init__.py 的目录。
myops/
├── __init__.py # 包初始化,可以写版本号
├── network/
│ ├── __init__.py
│ ├── ping.py
│ └── ssh.py
├── system/
│ ├── __init__.py
│ ├── disk.py
│ └── memory.py
└── utils/
├── __init__.py
└── logger.py
使用的时候:
from myops.network.ping import ping_host
from myops.system.disk import get_disk_usage
ping_host('8.8.8.8')
print(get_disk_usage('/'))
utils.py 塞一堆不相关的函数。我见过最离谱的 utils.py 有3000行,里面既有网络检测又有日志格式化还有时间转换……这种包,谁用谁想骂人。
4.3.3 模块的导入路径
Python 找模块的顺序是:
- 当前脚本所在目录
PYTHONPATH环境变量指定的路径- 系统标准库路径
- site-packages 第三方库路径
如果你自己写的包导入失败,八成是路径问题。我一般这样处理:
import sys
import os
# 把项目根目录加入路径
project_root = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, project_root)
# 现在可以正常导入了
from myops.network import ping
嗯,说白了,函数和模块就是让你的代码「可复用、可维护、可分享」。写一次,到处用——这才是自动化运维的精髓。你想想看,如果每次写脚本都从零开始,那跟手动操作有什么区别?