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

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

news 2026/6/16 18:21:11

从零到自动化：手把手教你用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.jsqmd.com/news/971883/

相关文章：

  • 洛帝牢垫圈应用场景有哪些 - myqiye
  • PyCharm远程解释器实战：用WSL2里的Conda环境跑通PyTorch GPU训练
  • 深度学习安全：权重扰动后门攻击与防御实战
  • 新手画板必看：我的PCB因为这几个接地错误，ESD测试直接挂了（附整改前后对比图）
  • 用联盛德HLK-W806和ST7567 LCD自制一个简易天气站：从驱动到UI显示的完整项目
  • IDEA条件断点进阶玩法：除了x＞21，还能用正则和脚本精准拦截线上Bug
  • 【26年面试题总结】构建生产级 Agent 系统:三个值得深挖的面试题
  • 从你家光猫到运营商机房：一趟PON（GPON/EPON）数据之旅的完整拆解
  • 电力仿真新手必看：用PSCAD搭建第一个RLC电路模型（附详细参数设置避坑点）
  • 2026年优质热敏条码打印机品牌排名，如何选择？ - myqiye
  • 用555定时器和CD4518做个复古电子钟：从原理图到面包板，手把手带你复刻数电课设
  • Pluto SDR玩转OFDM：除了频带利用率翻倍，我们还能用它做什么？
  • 从一次内存读写错误说起：深入理解C语言中size_t、uint64_t与long long的本质区别
  • 别再只用ArcMap了！深度解析ArcGIS Desktop三兄弟：ArcMap、ArcGlobe、ArcScene到底该怎么选？
  • 跑遍南山福田对比6家｜RERA激光封边，碾压传统EVA黑线脱胶 - 产品测评官
  • #深圳随机进店实测｜直击RERA工厂，揭秘85%转介绍率真相 - 产品测评官
  • 电力自动化工程师用的IEC61850 ICD文件快速生成与SCL可视化编辑工具
  • Claude Code 的 Skill 是什么？3 分钟看懂
  • 如何用WorkshopDL轻松下载Steam创意工坊模组？3步解决跨平台模组难题
  • HLK-W806驱动ST7567 LCD避坑指南：从初始化失败到完美显示的调试全记录
  • 公办二本认证院校有哪些？ - myqiye
  • 从游戏引擎到GIS：一文搞懂glTF与b3dm在Cesium 3D Tiles中的实战应用
  • MixIO平台保姆级入门：从零上手物联网项目（基于Mixly 2.0）
  • 保姆级教程：手把手教你用OBC4为不同总账科目组（如资产、负债）设置差异化的字段必填规则
  • Gemini3.0绑卡教程，全程无成本、无实体卡，快速完成
  • 5个步骤掌握MTKClient：拯救联发科设备的数据恢复神器
  • 告别枯燥理论：用NS-3.35手把手搭建你的第一个点对点网络仿真（附完整代码解析）
  • 告别FlexTimer！S32K3的eMIOS模块到底强在哪？保姆级配置流程分享
  • 2026年磁粉探伤机多少钱？射阳探伤机厂价格亲民 - myqiye
  • LeetCode 76 最小覆盖子串｜JS 滑动窗口标准解法（逐行精讲）