系统交互库
os 模块
os 模块提供了跨平台的文件和目录操作方法，是 python 标准库中最常用的系统接口之一：
文件操作：
os.rename(src, dst)- 重命名文件或目录os.remove(path)- 删除文件（不能删除目录）
os.path 子模块提供：
os.path.exists(path)- 检查路径是否存在os.path.isfile(path)/os.path.isdir(path)- 判断路径类型os.path.getsize(path)- 获取文件大小
目录操作：
os.mkdir(path)- 创建单个目录os.makedirs(path)- 递归创建多级目录os.listdir(path)- 列出目录内容os.chdir(path)- 改变当前工作目录os.walk(top)- 目录树生成器，递归遍历
路径处理：
os.path.join(path1, path2)- 智能拼接路径（跨平台兼容）os.path.abspath(path)- 获取绝对路径os.path.dirname(path)/os.path.basename(path)- 获取目录名/文件名os.path.splitext(path)- 分离文件名和扩展名
系统信息：
os.name- 操作系统名称（'posix', 'nt', 'java'）os.environ- 环境变量字典os.getcwd()- 获取当前工作目录os.cpu_count()- 获取cpu核心数
sys 模块
sys 模块实现了与 python 解释器的深度交互：
命令行参数：
sys.argv- 脚本参数列表（第一个元素是脚本名称）- 示例：
python script.py arg1 arg2→sys.argv = ['script.py', 'arg1', 'arg2']
模块搜索路径：
sys.path- python 模块搜索路径列表- 可动态修改：
sys.path.append('/custom/path')
标准输入输出：
sys.stdin/sys.stdout/sys.stderr- 标准 i/o 流对象- 重定向示例：
sys.stdout = open('output.log', 'w')
print("这将写入文件")
解释器信息：
sys.version- python 版本信息sys.platform- 操作系统平台标识符sys.executable- python 解释器路径sys.exit([status])- 退出程序（0表示成功）
其他重要功能：
sys.getsizeof(obj)- 获取对象占用的内存大小sys.getrecursionlimit()- 获取递归深度限制sys.settrace()- 设置跟踪函数用于调试
subprocess 模块
subprocess 提供了更安全的进程管理方式，替代传统的 os.system()：
基本用法：
# 运行简单命令
result = subprocess.run(["ls", "-l"], capture_output=true, text=true)
print(result.stdout)
# 检查返回码
if result.returncode != 0:
print("命令执行失败")
高级控制：
subprocess.popen()- 创建子进程并返回句柄
proc = subprocess.popen(["python", "worker.py"],
stdin=subprocess.pipe,
stdout=subprocess.pipe)
管道连接：
subprocess.pipe- 创建管道
p1 = subprocess.popen(["cat", "file.txt"], stdout=subprocess.pipe) p2 = subprocess.popen(["grep", "keyword"], stdin=p1.stdout, stdout=subprocess.pipe)
错误处理：
subprocess.calledprocesserror- 命令返回非零状态时抛出- 安全执行：
try:
subprocess.run(["rm", "important_file"], check=true)
except subprocess.calledprocesserror as e:
print(f"命令失败: {e}")
其他功能：
timeout参数 - 设置命令超时时间shell=true- 通过系统shell执行命令（有安全风险）env参数 - 设置子进程环境变量
远程管理工具
paramiko 库
paramiko 是纯 python 实现的 sshv2 协议库：
ssh 连接：
import paramiko
client = paramiko.sshclient()
client.set_missing_host_key_policy(paramiko.autoaddpolicy())
client.connect('hostname', username='user', password='pass')
执行命令：
stdin, stdout, stderr = client.exec_command('ls -l')
print(stdout.read().decode())
文件传输：
sftp = client.open_sftp()
sftp.put('local.txt', 'remote.txt') # 上传
sftp.get('remote.txt', 'local.txt') # 下载
sftp.close()
密钥认证：
key = paramiko.rsakey.from_private_key_file('/path/to/key.pem')
client.connect('host', username='user', pkey=key)
高级功能：
- ssh 隧道
- 保持连接
- 主机密钥验证
- 交互式 shell
fabric 工具
fabric2 是新一代的批量任务框架，构建在 paramiko 之上：
任务定义：
from fabric import task
@task
def deploy(c):
c.run('git pull')
c.run('pip install -r requirements.txt')
主机管理：
@task
def check_disk(c):
# 单个连接
result = c.run('df -h', hide=true)
print(f"磁盘使用:\n{result.stdout}")
# 多主机
for connection in c:
with connection.cd('/var/log'):
connection.run('ls -l')
批量执行：
from fabric import connection, threadinggroup
hosts = ['web1', 'web2', 'db1']
group = threadinggroup(*hosts, user='admin')
result = group.run('hostname')
for connection, result in result.items():
print(f"{connection.host}: {result.stdout.strip()}")
配置系统：
fabric.yml配置文件：
user: deploy connect_kwargs: key_filename: /path/to/key.pem hosts: - web1.example.com - web2.example.com
- 或通过 python 配置：
c = connection('host', user='user', connect_kwargs={'password': 'pass'})
其他特性：
- 并行执行
- 失败处理
- 上下文管理器
- 本地和远程操作混合
网络通信库
socket 模块
socket 提供底层网络接口：
tcp 通信：
# 服务器端
server_socket = socket.socket(socket.af_inet, socket.sock_stream)
server_socket.bind(('0.0.0.0', 8000))
server_socket.listen(1)
conn, addr = server_socket.accept()
# 客户端
client_socket = socket.socket(socket.af_inet, socket.sock_stream)
client_socket.connect(('localhost', 8000))
udp 通信：
# 服务器端
udp_socket = socket.socket(socket.af_inet, socket.sock_dgram)
udp_socket.bind(('0.0.0.0', 8000))
data, addr = udp_socket.recvfrom(1024)
# 客户端
udp_socket.sendto(b'hello', ('localhost', 8000))
常用方法：
socket.settimeout()- 设置超时时间socket.getaddrinfo()- 地址解析socket.gethostname()- 获取主机名socket.shutdown()- 优雅关闭连接
高级功能：
- 非阻塞 i/o
- 套接字选项设置
- 多路复用 (select/poll)
- ssl 包装
requests 库
requests 提供了人性化的 http 客户端：
简单请求：
# get 请求
response = requests.get('https://api.example.com/data')
print(response.json())
# post 请求
payload = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://api.example.com/post', data=payload)
参数传递：
params- url 查询参数
requests.get('https://api.example.com/search', params={'q': 'python'})
json- json 请求体
requests.post('https://api.example.com/data', json={'key': 'value'})
会话保持：
session = requests.session()
session.get('https://example.com/login', auth=('user', 'pass'))
response = session.get('https://example.com/dashboard')
高级功能：
- 代理支持：
proxies = {'http': 'http://proxy.example.com:8080'}
requests.get('http://example.com', proxies=proxies)
- 超时设置：
requests.get('http://example.com', timeout=5)
- ssl 验证：
requests.get('https://example.com', verify='/path/to/cert.pem')
响应处理：
response.status_code- http 状态码response.headers- 响应头response.content- 二进制响应体response.text- 文本响应体response.json()- json 解析
数据处理工具
json 模块
json 模块用于 api 数据交互：
序列化：
data = {'name': 'alice', 'age': 30, 'skills': ['python', 'sql']}
json_str = json.dumps(data, indent=2) # 美化输出
反序列化：
data = json.loads('{"name": "alice", "age": 30}')
文件操作：
# 写入文件
with open('data.json', 'w') as f:
json.dump(data, f)
# 读取文件
with open('data.json') as f:
data = json.load(f)
高级功能：
- 自定义编码器：
class customencoder(json.jsonencoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
json.dumps(obj, cls=customencoder)
- 解析参数：
json.loads(json_str, object_hook=my_decoder)
csv 模块
csv 模块用于报表生成：
读取 csv：
with open('data.csv') as f:
reader = csv.reader(f)
for row in reader:
print(row)
写入 csv：
with open('output.csv', 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['name', 'age'])
writer.writerow(['alice', 30])
字典格式：
# 读取
with open('data.csv') as f:
reader = csv.dictreader(f)
for row in reader:
print(row['name'], row['age'])
# 写入
with open('output.csv', 'w') as f:
fieldnames = ['name', 'age']
writer = csv.dictwriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow({'name': 'alice', 'age': 30})
方言支持：
csv.register_dialect('mydialect', delimiter='|', quoting=csv.quote_minimal)
with open('data.psv') as f:
reader = csv.reader(f, dialect='mydialect')
yaml 模块
yaml 用于配置文件解析：
基本用法：
import yaml
# 解析 yaml
with open('config.yaml') as f:
config = yaml.safe_load(f)
# 生成 yaml
data = {'server': {'host': '127.0.0.1', 'port': 8000}}
yaml_str = yaml.dump(data, default_flow_style=false)
高级特性：
- 锚点和别名：
defaults: &defaults adapter: postgres host: localhost development: <<: *defaults database: dev
- 自定义标签
- 多文档流
安全考虑：
- 始终优先使用
yaml.safe_load()而非yaml.load() - 避免加载不受信任的 yaml 文件
- 复杂对象需要自定义表示器/构造器
与其他格式对比：
- 比 json 更易读
- 支持注释
- 比 xml 更简洁
- 适合配置文件和复杂数据结构
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持代码网。