当前位置： 首页 > news >正文

从零到自动化：手把手教你用Python脚本调用Redfish API管理服务器（附Postman转Python代码技巧）

news 2026/6/8 2:30:59

从零到自动化：手把手教你用Python脚本调用Redfish API管理服务器（附Postman转Python代码技巧）

当你已经熟悉了在Postman中测试Redfish接口的基本操作，却苦于无法将这些零散的请求整合成自动化流程时，这篇文章正是为你准备的。我们将从Postman的测试场景出发，逐步构建一个完整的Python自动化框架，让你能够轻松管理服务器生命周期中的各种操作。

Redfish作为现代服务器管理的标准接口，其RESTful设计让自动化变得可能。但要从手动测试转向生产级脚本，我们需要解决认证管理、错误处理、并发控制等一系列问题。下面，我将分享如何将Postman中的12个典型用例转化为可靠的Python实现。

1. 环境准备与基础架构

在开始编码前，我们需要搭建好Python环境并理解Redfish的基本工作流程。不同于Postman的一次性测试，自动化脚本需要考虑长期运行的稳定性。

1.1 安装必要依赖

pip install requests urllib3

建议使用虚拟环境隔离项目依赖。对于生产环境，还可以考虑添加retrying库用于错误重试：

pip install retrying

1.2 构建基础请求类

我们将创建一个RedfishClient类来封装所有公共操作，避免重复代码。这个类需要处理：

  • 会话管理（认证令牌获取与刷新）
  • 请求重试机制
  • 统一的错误处理
  • 响应解析
import requests from urllib3.exceptions import InsecureRequestWarning # 禁用SSL警告（仅用于测试环境） requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) class RedfishClient: def __init__(self, base_url, username, password): self.base_url = base_url.rstrip('/') self.session = requests.Session() self.session.verify = False # 生产环境应配置合法证书 self.auth_token = None self._login(username, password) def _login(self, username, password): url = f"{self.base_url}/redfish/v1/SessionService/Sessions" headers = {'Content-Type': 'application/json'} payload = {'UserName': username, 'Password': password} response = self.session.post(url, json=payload, headers=headers) response.raise_for_status() self.auth_token = response.headers.get('X-Auth-Token') self.session.headers.update({'X-Auth-Token': self.auth_token})

注意：生产环境务必配置正确的SSL证书验证，禁用验证仅适用于测试环境

2. 核心功能实现

现在我们可以基于这个基础类，实现各种服务器管理操作。让我们从最常用的几个功能开始。

2.1 服务器电源管理

电源控制是服务器管理中最基础也最重要的操作。Redfish提供了标准的Reset动作类型：

class RedfishClient: # ... 延续上面的类定义 def power_control(self, system_id=1, action='On'): """ 控制服务器电源状态 :param system_id: 系统ID，默认为1 :param action: 操作类型 ['On', 'GracefulShutdown', 'ForceRestart'] """ valid_actions = ['On', 'GracefulShutdown', 'ForceRestart'] if action not in valid_actions: raise ValueError(f"Invalid action. Must be one of {valid_actions}") url = f"{self.base_url}/redfish/v1/Systems/{system_id}/Actions/ComputerSystem.Reset" payload = {'ResetType': action} response = self.session.post(url, json=payload) response.raise_for_status() return response.json()

使用示例：

client = RedfishClient('https://192.168.1.100', 'admin', 'password') # 正常关机 client.power_control(action='GracefulShutdown') # 开机 client.power_control(action='On')

2.2 资产管理信息获取

获取服务器硬件信息是监控和资产管理的基础。Redfish的Systems端点提供了丰富的资产数据：

def get_system_info(self, system_id=1): """获取服务器详细信息""" url = f"{self.base_url}/redfish/v1/Systems/{system_id}" response = self.session.get(url) response.raise_for_status() data = response.json() return { 'model': data.get('Model'), 'manufacturer': data.get('Manufacturer'), 'serial_number': data.get('SerialNumber'), 'power_state': data.get('PowerState'), 'bios_version': data.get('BiosVersion'), 'processor_summary': data.get('ProcessorSummary', {}).get('Count'), 'memory_summary': data.get('MemorySummary', {}).get('TotalSystemMemoryGiB') }

3. 用户账户管理自动化

BMC用户管理是运维中的重要工作，特别是当需要批量部署或定期更换凭证时。

3.1 创建用户账户

def create_user(self, user_id, username, password, role='Administrator'): """创建新的BMC用户""" url = f"{self.base_url}/redfish/v1/AccountService/Accounts" payload = { 'Id': str(user_id), 'UserName': username, 'Password': password, 'RoleId': role } response = self.session.post(url, json=payload) if response.status_code == 400: error = response.json().get('error', {}) if 'already exists' in error.get('message', ''): raise ValueError(f"User ID {user_id} already exists") response.raise_for_status() return response.json()

3.2 安全更新用户信息

Redfish要求使用ETag机制来防止并发修改冲突，这与Postman中的If-Match头对应：

def update_user(self, user_id, new_username=None, new_password=None, role=None): """更新用户信息，需要先获取ETag""" # 先获取当前用户信息和ETag info_url = f"{self.base_url}/redfish/v1/AccountService/Accounts/{user_id}" info_resp = self.session.get(info_url) info_resp.raise_for_status() etag = info_resp.headers.get('ETag') if not etag: raise RuntimeError("ETag not found in response headers") # 准备更新数据 payload = {} if new_username: payload['UserName'] = new_username if new_password: payload['Password'] = new_password if role: payload['RoleId'] = role if not payload: raise ValueError("No update fields provided") # 执行更新 headers = {'If-Match': etag} update_url = f"{self.base_url}/redfish/v1/AccountService/Accounts/{user_id}" response = self.session.patch(update_url, json=payload, headers=headers) response.raise_for_status() return response.json()

4. 网络配置与BIOS设置

4.1 修改BMC网络配置

def update_bmc_network(self, interface_id, ipv4_address=None, ipv6_address=None): """修改BMC管理网络配置""" # 获取当前接口信息和ETag url = f"{self.base_url}/redfish/v1/Managers/1/EthernetInterfaces/{interface_id}" info_resp = self.session.get(url) info_resp.raise_for_status() etag = info_resp.headers.get('ETag') if not etag: raise RuntimeError("ETag not found in response headers") # 准备更新数据 payload = {} if ipv4_address: payload['IPv4Addresses'] = [{'Address': ipv4_address}] if ipv6_address: payload['IPv6Addresses'] = [{'Address': ipv6_address}] headers = {'If-Match': etag} response = self.session.patch(url, json=payload, headers=headers) response.raise_for_status() return response.json()

4.2 修改BIOS启动顺序

def update_bios_boot_order(self, boot_order): """ 修改BIOS启动顺序 :param boot_order: 启动顺序列表，如 ['HardDiskDrive', 'DVDROMDrive', 'PXE'] """ # 获取当前BIOS设置和ETag url = f"{self.base_url}/redfish/v1/Systems/1/Bios/Settings" info_resp = self.session.get(url) info_resp.raise_for_status() etag = info_resp.headers.get('ETag') if not etag: raise RuntimeError("ETag not found in response headers") # 构建启动顺序属性 attributes = {} for i, device in enumerate(boot_order): attributes[f'BootTypeOrder{i}'] = device payload = {'Attributes': attributes} headers = {'If-Match': etag} response = self.session.patch(url, json=payload, headers=headers) response.raise_for_status() return response.json()

5. 从Postman到Python的转换技巧

Postman提供了方便的"Generate Code Snippet"功能，可以快速将请求转换为Python代码。但直接使用生成的代码往往不够健壮，我们需要进行优化。

5.1 利用Postman生成代码骨架

  1. 在Postman中构造好请求并测试通过
  2. 点击右侧的"Code"按钮
  3. 选择"Python - Requests"语言
  4. 复制生成的代码片段

生成的代码示例：

import requests url = "https://{{deviceip}}/redfish/v1/SessionService/Sessions" payload = { "UserName": "用户名", "Password": "密码" } headers = { "Content-Type": "application/json" } response = requests.request("POST", url, json=payload, headers=headers, verify=False) print(response.text)

5.2 优化生成的代码

直接生成的代码有几个问题需要改进：

  1. 硬编码的凭证信息
  2. 缺乏错误处理
  3. 没有会话复用
  4. 忽略重要的响应头信息

优化后的版本：

def login(base_url, username, password): """优化的登录函数""" url = f"{base_url}/redfish/v1/SessionService/Sessions" headers = {'Content-Type': 'application/json'} payload = {'UserName': username, 'Password': password} try: response = requests.post( url, json=payload, headers=headers, verify=False, # 生产环境应配置正确证书 timeout=10 ) response.raise_for_status() auth_token = response.headers.get('X-Auth-Token') if not auth_token: raise ValueError("X-Auth-Token not found in response headers") return auth_token except requests.exceptions.RequestException as e: raise RuntimeError(f"Login failed: {str(e)}")

6. 错误处理与健壮性设计

生产环境的自动化脚本必须具备完善的错误处理机制。以下是几个关键点：

6.1 重试机制

对于临时性网络问题或服务短暂不可用，合理的重试可以提高成功率：

from retrying import retry class RedfishClient: @retry(stop_max_attempt_number=3, wait_fixed=2000) def _make_request(self, method, path, **kwargs): """带重试的请求封装""" url = f"{self.base_url}{path}" try: response = self.session.request(method, url, **kwargs) if response.status_code >= 500: raise requests.exceptions.RequestException( f"Server error: {response.status_code}" ) return response except requests.exceptions.ConnectionError: raise except requests.exceptions.RequestException as e: if self._should_retry(e): raise raise

6.2 统一错误处理

定义自定义异常类，提供更清晰的错误信息：

class RedfishError(Exception): """Redfish操作异常基类""" pass class AuthenticationError(RedfishError): """认证失败异常""" pass class ResourceNotFoundError(RedfishError): """资源不存在异常""" pass class ConcurrentModificationError(RedfishError): """并发修改冲突异常""" pass

然后在各方法中抛出适当的异常：

def get_system_info(self, system_id=1): try: response = self._make_request('GET', f'/redfish/v1/Systems/{system_id}') if response.status_code == 404: raise ResourceNotFoundError(f"System {system_id} not found") response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: raise RedfishError(f"Failed to get system info: {str(e)}")

7. 进阶技巧与最佳实践

7.1 并发操作处理

当多个系统同时尝试修改同一资源时，ETag机制可以防止数据竞争。以下是处理并发修改的完整流程：

def safe_update_resource(self, resource_path, update_func, max_retries=3): """ 安全更新资源，自动处理ETag冲突 :param resource_path: 资源路径，如 '/redfish/v1/Systems/1' :param update_func: 接收当前资源数据，返回更新内容的函数 :param max_retries: 最大重试次数 """ for attempt in range(max_retries): # 获取当前资源状态 get_response = self._make_request('GET', resource_path) if get_response.status_code == 404: raise ResourceNotFoundError(f"Resource {resource_path} not found") get_response.raise_for_status() current_data = get_response.json() etag = get_response.headers.get('ETag') if not etag: raise RuntimeError("ETag not found in response headers") # 准备更新内容 update_data = update_func(current_data) if not update_data: return current_data # 没有需要更新的内容 # 尝试更新 headers = {'If-Match': etag} try: patch_response = self._make_request( 'PATCH', resource_path, json=update_data, headers=headers ) patch_response.raise_for_status() return patch_response.json() except requests.exceptions.HTTPError as e: if e.response.status_code == 412: # Precondition Failed if attempt < max_retries - 1: continue # 重试 raise ConcurrentModificationError( f"Failed to update after {max_retries} attempts due to concurrent modifications" ) raise raise ConcurrentModificationError("Max retries reached for concurrent update")

使用示例：

def update_boot_order(current): new_order = ['HardDiskDrive', 'DVDROMDrive', 'PXE'] attributes = {} for i, device in enumerate(new_order): attributes[f'BootTypeOrder{i}'] = device return {'Attributes': attributes} client.safe_update_resource('/redfish/v1/Systems/1/Bios/Settings', update_boot_order)

7.2 批量操作与任务编排

对于需要按顺序执行的多个操作，可以创建任务链：

def perform_server_maintenance(self, system_id=1): """执行服务器维护任务序列""" operations = [ {'name': 'GracefulShutdown', 'func': self.power_control, 'args': {'action': 'GracefulShutdown'}}, {'name': 'UpdateBIOS', 'func': self.update_bios_settings, 'args': {}}, {'name': 'PowerOn', 'func': self.power_control, 'args': {'action': 'On'}} ] results = [] for op in operations: try: result = op['func'](**op['args']) results.append({ 'operation': op['name'], 'status': 'success', 'result': result }) except RedfishError as e: results.append({ 'operation': op['name'], 'status': 'failed', 'error': str(e) }) break # 关键路径失败则中止 return results

7.3 日志记录与审计

完善的日志记录对于自动化运维至关重要：

import logging class RedfishClient: def __init__(self, base_url, username, password): self.logger = logging.getLogger('RedfishClient') # ... 其他初始化代码 def _log_request(self, method, url, response=None, error=None): """记录请求日志""" log_data = { 'method': method, 'url': url, 'status': getattr(response, 'status_code', None) if response else None, 'error': str(error) if error else None } if error: self.logger.error("Request failed", extra=log_data) else: self.logger.info("Request completed", extra=log_data) def _make_request(self, method, path, **kwargs): """带日志记录的请求封装""" url = f"{self.base_url}{path}" try: response = self.session.request(method, url, **kwargs) self._log_request(method, url, response) return response except Exception as e: self._log_request(method, url, error=e) raise

配置日志格式：

logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('redfish_operations.log'), logging.StreamHandler() ] )
http://www.cnnetsun.cn/news/2816018.html

相关文章：

  • Pluto SDR新手避坑指南：搞定MATLAB驱动配置，快速搭建你的第一个无线收发链路
  • 告别枯燥理论：用NS-3.35手把手搭建你的第一个点对点网络仿真（附完整代码解析）
  • 模板驱动文档自动化：告别重复劳动的确定性交付方案
  • 用CODESYS ST语言给官方梯形图教程写个仿真，我发现了这些设计细节
  • 哔哩下载姬DownKyi：5分钟掌握B站视频批量下载的终极指南
  • 音频处理实战：用Python快速设计Butterworth滤波器并可视化幅频曲线（附Jupyter Notebook）
  • 别再手动解压了！用Docker在Linux服务器上5分钟部署Matlab 2018b运行环境
  • AD9361接收链路调试踩坑记：从官方配置软件到SPI寄存器，手把手教你避开ENSM状态这个‘大坑’
  • 世界卫生大会健康中国建设 大健康医药产业理论体系数智化健康服务
  • JavaSE 和 JavaEE 是什么意思
  • TOPSIS、AHP、熵权法怎么选？三大决策分析模型对比与避坑指南
  • 别再死记叉乘公式了！用Python和NumPy玩转向量运算与反对称矩阵
  • ESP32 AT固件Web Captive Portal避坑指南：为什么你的热点SSID必须叫‘pos_softap’？
  • C语言指针之二malloc的用法及详解
  • 单人创业，靠 StarLny 搭建数字团队
  • 避坑指南：ABAP里同时调用WS_REVERSE_GOODS_ISSUE和BAPI_OUTB_DELIVERY_CHANGE报VL216错误的深层原因与替代方案
  • Infra CONVERT 德国标准下的图纸自动化识别与检验计划生成指南
  • 完全免费的Android开源相机神器：OpenCamera专业摄影指南
  • 【stack、queue、deque、priority_queue】C++ 栈 / 队列 / 优先级队列全解析！手撕实现 + 二叉树层序遍历（附源码）
  • KMS_VL_ALL_AIO：Windows与Office批量激活的终极技术方案
  • 保姆级教程：用FNL数据从零搭建WRF环境并成功运行第一个案例（避坑指南）
  • 告别phpMyAdmin！一个Docker容器搞定MySQL、PostgreSQL、MongoDB，Adminer保姆级安装与多数据库连接实战
  • Windows 10/11 下用 Visual Studio 2019 编译 ZLMediaKit 流媒体服务，保姆级避坑指南
  • 信号处理实战：用db4小波分析你的传感器数据（MATLAB验证+C语言移植指南）
  • AI人脸识别考勤签到系统
  • 别再手动整理BOM了！用Excel自定义Altium Designer料单模板，效率翻倍（附模板文件）
  • 【闲聊】孩子越长大为什么越不愿意和父母讲心里话（亿点不一样）
  • 第【7】期--自由空间光通信(FSO)在Gamma-Gamma湍流信道下的BER性能仿真-maltab完整代码+报告
  • 零基础落地！三个精益实操技巧，激活员工主动改善意识
  • 别再死记硬背了！一张图+Python脚本帮你彻底搞懂ISO15765-2网络层多帧传输与流控