影刀RPA Python节点调用API的完整模板：GET POST PUT DELETE全家桶

影刀RPA Python节点调用API的完整模板GET POST PUT DELETE全家桶影刀的【HTTP请求】动作能覆盖简单的API调用场景。但一旦涉及到Bearer Token认证、JSON复杂嵌套、文件上传、流式响应——Python的requests库才是正解。这篇文章给出一个在影刀Python节点中调用API的标准模板覆盖GET/POST/PUT/DELETE四种方法。基础模板importrequestsimportjsonclassAPIClient:def__init__(self,base_url,tokenNone):self.base_urlbase_url.rstrip(/)self.sessionrequests.Session()# 默认请求头self.session.headers.update({User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36,Accept:application/json,Content-Type:application/json,})iftoken:self.session.headers[Authorization]fBearer{token}def_request(self,method,path,**kwargs):统一请求方法带重试和错误处理urlf{self.base_url}{path}max_retries3forattemptinrange(1,max_retries1):try:responseself.session.request(method,url,timeout30,**kwargs)response.raise_for_status()# 尝试解析JSONtry:returnresponse.json()except:returnresponse.textexceptrequests.Timeout:print(f请求超时{attempt}/{max_retries}{method}{url})exceptrequests.ConnectionError:print(f连接失败{attempt}/{max_retries}{method}{url})exceptrequests.HTTPErrorase:statuse.response.status_codeifstatusin(401,403):print(f认证失败{status})raise# 认证问题不重试elifstatus500:print(f服务器错误{attempt}/{max_retries}{status})else:print(f客户端错误{status})try:print(f 响应{e.response.json()})except:print(f 响应{e.response.text})raise# 4xx通常不是暂时的ifattemptmax_retries:importtime time.sleep(2**attempt)# 指数退避raiseException(f请求失败已重试{max_retries}次)defget(self,path,paramsNone):returnself._request(GET,path,paramsparams)defpost(self,path,dataNone):returnself._request(POST,path,jsondata)defput(self,path,dataNone):returnself._request(PUT,path,jsondata)defdelete(self,path):returnself._request(DELETE,path)# 使用示例 # 初始化客户端apiAPIClient(base_urlhttps://api.example.com/v1,tokenyour_bearer_token_here)# GET请求查询列表ordersapi.get(/orders,params{page:1,size:50})print(f获取到{len(orders.get(data,[]))}条订单)# POST请求创建记录new_orderapi.post(/orders,data{customer:张三,amount:3580.50,items:[{sku:A001,qty:2},{sku:B002,qty:1},]})print(f创建成功ID{new_order[id]})# PUT请求更新记录updatedapi.put(/orders/123,data{status:已发货})print(f更新成功{updated[status]})# DELETE请求api.delete(/orders/456)print(删除成功)认证方式店群矩阵自动化突破运营极限Token认证Bearerheaders{Authorization:fBearer{token}}Basic认证fromrequests.authimportHTTPBasicAuth responserequests.get(url,authHTTPBasicAuth(username,password))API Key自定义Headerheaders{X-API-Key:your_api_key_here}SessionCookie认证sessionrequests.Session()# 先登录login_respsession.post(https://api.example.com/login,json{username:admin,password:pass123})# 后续请求自动携带Cookiedatasession.get(https://api.example.com/data)文件上传# 上传单个文件withopen(rC:\data\report.xlsx,rb)asf:responserequests.post(https://api.example.com/upload,files{file:(report.xlsx,f,application/vnd.openxmlformats-officedocument.spreadsheetml.sheet)})# 上传多个文件 额外参数files[(files,(file1.xlsx,open(file1.xlsx,rb))),(files,(file2.xlsx,open(file2.xlsx,rb))),]data{folder:reports,overwrite:true}requests.post(https://api.example.com/upload,filesfiles,datadata)分页获取全部数据deffetch_all_pages(api,path,page_size100):自动翻页获取全部数据all_data[]page1whileTrue:responseapi.get(path,params{page:page,size:page_size})itemsresponse.get(data,[])ifnotitems:breakall_data.extend(items)print(f第{page}页{len(items)}条累计{len(all_data)}条)# 判断是否还有下一页totalresponse.get(total,0)iflen(all_data)total:breakpage1print(f全部获取完成共{len(all_data)}条)returnall_datatemu店群自动化报活动案例错误码处理defhandle_api_error(response):统一处理API错误statusresponse.status_codeifstatus401:return认证失败Token可能过期elifstatus403:return权限不足elifstatus404:return资源不存在elifstatus429:retry_afterresponse.headers.get(Retry-After,60)returnf请求频率过高{retry_after}秒后可重试elifstatus500:return服务器内部错误建议稍后重试else:returnf未知错误{status}在影刀中的完整调用链影刀流程中的API调用 1. Python节点 → 加载配置API URL、Token 2. Python节点 → 调用API获取数据 3. 返回结果给影刀变量 4. 影刀处理结果写入Excel/数据库/发送通知示例【读取全局变量】→ api_token Python节点获取订单列表 api APIClient(https://api.example.com/v1, tokenapi_token) orders api.get(/orders, params{date: 2026-07-01}) → 输出变量orders_data  【ForEach】→ 遍历 orders_data 处理每条订单... END FOR总结Python的requests库是API调用的瑞士军刀。把重试、超时、错误处理都封装在_request方法里业务代码只关心api.get(/xxx)和返回的数据。这个模板放到你的项目里以后API调用全部用它。作者林焱