python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python系统交互库

Python系统交互库解析过程

作者:Sally璐璐

本文介绍了Python中os、sys、subprocess等模块及paramiko、Fabric、socket、requests等库的核心功能,涵盖文件操作、系统交互、进程管理、网络通信、数据处理等场景,并强调了安全使用YAML等格式的注意事项

系统交互库

os 模块

os 模块提供了跨平台的文件和目录操作方法,是 Python 标准库中最常用的系统接口之一:

文件操作

os.path 子模块提供:

目录操作

路径处理

系统信息

sys 模块

sys 模块实现了与 Python 解释器的深度交互:

命令行参数

模块搜索路径

标准输入输出

sys.stdout = open('output.log', 'w')
print("这将写入文件")

解释器信息

其他重要功能

subprocess 模块

subprocess 提供了更安全的进程管理方式,替代传统的 os.system():

基本用法

# 运行简单命令
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(result.stdout)

# 检查返回码
if result.returncode != 0:
    print("命令执行失败")

高级控制

proc = subprocess.Popen(["python", "worker.py"], 
                       stdin=subprocess.PIPE,
                       stdout=subprocess.PIPE)

管道连接

p1 = subprocess.Popen(["cat", "file.txt"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", "keyword"], stdin=p1.stdout, stdout=subprocess.PIPE)

错误处理

try:
    subprocess.run(["rm", "important_file"], check=True)
except subprocess.CalledProcessError as e:
    print(f"命令失败: {e}")

其他功能

远程管理工具

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)

高级功能

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()}")

配置系统

user: deploy
connect_kwargs:
  key_filename: /path/to/key.pem
hosts:
  - web1.example.com
  - web2.example.com
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))

常用方法

高级功能

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)

参数传递

requests.get('https://api.example.com/search', params={'q': 'python'})
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)
requests.get('https://example.com', verify='/path/to/cert.pem')

响应处理

数据处理工具

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

安全考虑

与其他格式对比

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文