Python+Requests 企业级接口测试入门
1、GET接口简单案例
我的答案:
import time
import requestsdef get_all_users():try:response = requests.get(url='https://reqres.in/api/users', #接口地址timeout=10 #超时时间)response.raise_for_status() #状态#return response.json() #返回接口数据 # 返回带性能指标的数据data = response.json() #data获取接口数据data['metadata'] = { 'response_time_ms': response.elapsed.total_seconds() * 1000, #加了反应时间和状态码'status_code': response.status_code}return dataexcept Exception as e:print(f'异常问题包含:{str(e)}')if __name__ == "__main__":start_time = time.time()result = get_all_users()end_time = time.time()elapsed_time = end_time - start_timeprint(f'请求接口耗时:{elapsed_time:.2f}s')if result:print(f"API返回数据: {result}")print(f"服务端响应时间: {result['metadata']['response_time_ms']:.2f}ms")
老师点评:
做得好的地方
- HTTPS加密:正确使用
https
协议 - 超时控制:设置了合理的10秒超时
- 异常处理:捕获了所有可能的异常(但建议细化异常类型,见改进建议)
- JSON解析:使用
.json()
而非.text
返回结构化数据 - 耗时统计:添加了性能监控逻辑(符合APM监控要求)
优化点
1. 异常处理精细化
except requests.exceptions.HTTPError as http_err:logging.error(f"HTTP错误: {http_err}") # 比如403/404
except requests.exceptions.Timeout:logging.warning("请求超时,请检查网络或重试")
except requests.exceptions.RequestException as err:logging.critical(f"致命错误: {str(err)}")
2. 添加重试机制(生产环境必备)
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retrysession = requests.Session()
retries = Retry(total=3, # 最大重试次数backoff_factor=1, # 重试间隔(秒)status_forcelist=[500, 502, 503, 504] # 遇到这些状态码才重试
)
session.mount('https://', HTTPAdapter(max_retries=retries))
3. 安全增强
response = requests.get(url='https://reqres.in/api/users',timeout=(3.05, 10), # 分别设置连接/读取超时verify=True # 生产环境必须验证SSL证书
)
4. 性能优化建议
# 在返回前添加响应时间记录(更专业的方式)
data = response.json()
data['response_time_ms'] = response.elapsed.total_seconds() * 1000
return data
📝 修改后的完整代码
import requests
import time
import logging
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retrylogging.basicConfig(level=logging.INFO)def get_all_users():session = requests.Session()# 配置自动重试retries = Retry(total=3,backoff_factor=1,status_forcelist=[500, 502, 503, 504])session.mount('https://', HTTPAdapter(max_retries=retries))try:response = session.get(url='https://reqres.in/api/users',timeout=(3.05, 10), # (连接超时, 读取超时)verify=True)response.raise_for_status()# 返回带性能指标的数据data = response.json()data['metadata'] = {'response_time_ms': response.elapsed.total_seconds() * 1000,'status_code': response.status_code}return dataexcept requests.exceptions.HTTPError as http_err:logging.error(f"HTTP错误: {http_err}")except requests.exceptions.Timeout:logging.warning("请求超时,请检查网络或重试")except requests.exceptions.RequestException as err:logging.critical(f"请求失败: {str(err)}")return Noneif __name__ == "__main__":start_time = time.time()result = get_all_users()elapsed_time = (time.time() - start_time) * 1000 # 转为毫秒if result:print(f"API返回数据: {result}")print(f"服务端响应时间: {result['metadata']['response_time_ms']:.2f}ms")print(f"总耗时: {elapsed_time:.2f}ms")
2、post接口简单案例
我的思路:
-
异常处理
-
换GET为post
-
post与GET不同的点,GET直接使用参数,如params={‘key’: f"{api_key}“,‘location’:f”{city}",‘language’:“zh-Hans”,‘unit’:‘c’},post用的是结构体把数据框起来,例如json=
-
{'product_ids':product_ids'address':address }
还有headers
我的答案:
import hashlibimport requestsURL = "https://example.com/aip/login"
# 脚手架代码
def login(username, password):"""返回示例:{'token': 'abc123', 'user_id': 1001}"""encrypted_passwd = hashlib.md5(password.encode()).hexdigest()logindata={'usename' : username,'password' : encrypted_passwd}try:respone = requests.post(url=f"{URL}/login",json= logindata,)return respone.json()['token']except Exception as e:print("登录失败")def create_order(token, product_ids: list, address: dict):# 你的代码实现处try:headers = {'Authorization' : token,"Content-Type": "application/json"}data = {'product_ids':product_ids'address':address}response = requests.post(f"{URL}/orders",json=data,headers=headers)response.raise_for_status()return response.json()except requests.exceptions.HTTPError as httperr:print(f"接口报错:{httperr}")if {httperr} == 401:print("库存不足")except requests.exceptions.Timeout:print("请求超时")except requests.exceptions.RequestException as err:print(f"请求失败:{err}")if __name__ == "__main__":result = login("1001", "ShoIsRxywozHNaBz0")create_order(result,'[1,2,3]',"一串地址")
老师点评:
代码问题诊断
1. 基础语法错误
# 问题行(缺少逗号)
data = {'product_ids': product_ids # 这里缺少逗号'address': address
}
2. 安全风险
# 明文打印敏感信息(token)
print(f"登录token: {result}") # 实际代码中应避免
3. 逻辑缺陷
# 库存不足判断错误({httperr}是对象不是状态码)
if {httperr} == 401: # 错误写法
# 正确应该:
if response.status_code == 4001:
4. 类型不一致
# 参数类型声明为list但传入字符串
create_order(result, '[1,2,3]', ...) # 应该是 [1,2,3]
🚀 优化后的企业级代码
python
复制
import hashlib
import requests
from typing import Dict, List, Optional
import logging# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s'
)class OrderSystem:BASE_URL = "https://api.example.com" # 建议用类变量管理URL@staticmethoddef _encrypt_password(password: str) -> str:"""密码加密处理(符合企业安全规范)"""return hashlib.md5(password.encode()).hexdigest()@staticmethoddef login(username: str, password: str) -> Optional[str]:"""安全登录实现"""try:response = requests.post(url=f"{OrderSystem.BASE_URL}/login",json={'username': username, # 修正字段名usename -> username'password': OrderSystem._encrypt_password(password)},timeout=5,verify=True # 生产环境必须开启SSL验证)response.raise_for_status()return response.json().get('token')except requests.exceptions.HTTPError as e:logging.error(f"登录失败 HTTP {e.response.status_code}")except Exception as e:logging.error(f"登录异常: {str(e)}")return None@staticmethoddef create_order(token: str, product_ids: List[int], address: Dict) -> Dict:"""创建订单(带完整异常处理和日志)"""try:response = requests.post(url=f"{OrderSystem.BASE_URL}/orders",headers={'Authorization': f"Bearer {token}",'Content-Type': 'application/json'},json={'product_ids': product_ids,'address': address},timeout=10)# 优先检查业务状态码(企业系统常见设计)if response.json().get('code') == 4001:logging.warning("库存不足")return {'status': 'failed', 'reason': 'out_of_stock'}response.raise_for_status()return response.json()except requests.exceptions.HTTPError as e:logging.error(f"订单创建失败: HTTP {e.response.status_code}")except Exception as e:logging.error(f"订单创建异常: {str(e)}")return {'status': 'error'}if __name__ == "__main__":# 测试用例token = OrderSystem.login("test_user", "mypassword123")if token:result = OrderSystem.create_order(token=token,product_ids=[1, 2, 3], # 修正为正确list类型address={'city': '北京','street': '朝阳区某某路','postcode': '100000'})logging.info(f"订单创建结果: {result}")
✅ 企业级改进点
-
类型注解完善
- 所有方法添加了
-> Optional[str]
等返回类型声明
- 所有方法添加了
-
安全增强
- 密码加密独立成方法
_encrypt_password
- 使用
Bearer Token
标准鉴权方式
- 密码加密独立成方法
-
日志标准化
- 使用logging模块分级记录(INFO/WARNING/ERROR)
-
业务状态码处理
python
复制
if response.json().get('code') == 4001: # 优先检查业务自定义状态码
-
代码可测试性
- 将核心逻辑封装为静态方法,便于单元测试mock