功能简介
HttpRequest 类主要提供以下功能：
- 发送 GET 请求：用于从服务器获取资源。
- 发送 POST 请求：用于向服务器发送数据。
- 设置请求头：可以设置自定义的请求头，进行身份验证、内容类型设置等。
- 设置查询参数：用于向服务器传递查询参数，适用于 GET 请求。
- 表单数据提交：通过 POST 请求提交表单数据。
- 发送 JSON 数据：通过 POST 请求发送 JSON 格式的数据。
- 文件上传：通过 POST 请求上传文件到服务器。
- 文件下载：通过 GET 请求下载文件并保存到本地。
- 获取响应状态码：获取响应的 HTTP 状态码。
- 获取响应头：获取服务器返回的响应头。
- 基本认证：支持 HTTP 基本认证（用户名和密码）。
- 禁用 SSL 证书验证：禁用 SSL 证书验证，适用于开发和测试环境。
import requests # 导入requests库，用于发送HTTP请求
from requests.exceptions import Timeout, ConnectionError # 导入可能出现的异常
from requests.auth import HTTPBasicAuth # 导入HTTP基本认证模块
class HttpRequest:
def __init__(self):
self.session = requests.Session() # 创建一个会话对象，用于保持连接和配置全局设置
def send_get_request(self, url, headers=None, params=None, timeout=None):
"""
发送GET请求
"""
try:
response = self.session.get(url, headers=headers, params=params, timeout=timeout) # 发送GET请求并等待响应
return response.text # 返回响应的文本内容
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获超时和连接错误，打印错误信息
return None # 请求失败返回None
def send_post_request(self, url, headers=None, data=None, json=None, timeout=None):
"""
发送POST请求
"""
try:
response = self.session.post(url, headers=headers, data=data, json=json, timeout=timeout) # 发送POST请求
return response.text # 返回响应文本
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获异常并输出错误信息
return None # 请求失败返回None
def set_custom_headers(self, headers):
"""
设置自定义请求头
"""
self.session.headers.update(headers) # 更新session中的默认请求头为自定义的头部
def set_query_params(self, params):
"""
设置查询参数
"""
self.session.params = params # 设置查询参数，可以直接在GET请求中使用
def send_form_data(self, url, data=None, headers=None, timeout=None):
"""
发送表单数据
"""
try:
response = self.session.post(url, data=data, headers=headers, timeout=timeout) # 发送表单数据的POST请求
return response.text # 返回响应的文本内容
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获异常并打印错误信息
return None # 请求失败返回None
def send_json_data(self, url, json=None, headers=None, timeout=None):
"""
发送JSON数据
"""
try:
response = self.session.post(url, json=json, headers=headers, timeout=timeout) # 发送JSON数据的POST请求
return response.text # 返回响应的文本内容
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获异常并输出错误信息
return None # 请求失败返回None
def upload_file(self, url, file=None, headers=None, timeout=None):
"""
上传文件
"""
try:
files = {'file': file} # 使用字典封装文件
response = self.session.post(url, files=files, headers=headers, timeout=timeout) # 发送带文件的POST请求
return response.text # 返回响应的文本内容
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获异常并输出错误信息
return None # 请求失败返回None
def download_file(self, url, save_as=None, headers=None, timeout=None):
"""
下载文件
"""
try:
response = self.session.get(url, headers=headers, timeout=timeout) # 发送GET请求以下载文件
if save_as: # 如果指定了保存路径
with open(save_as, 'wb') as f: # 以二进制写模式打开文件
f.write(response.content) # 将响应内容写入文件
return response.text # 返回响应文本
except (Timeout, ConnectionError, IOError) as e:
print(f'Request failed: {e}') # 捕获网络或文件操作异常并打印错误信息
return None # 请求失败返回None
def get_response_status_code(self, url, headers=None, params=None, timeout=None):
"""
获取响应状态码
"""
try:
response = self.session.get(url, headers=headers, params=params, timeout=timeout) # 发送GET请求并获取响应
return response.status_code # 返回响应的状态码
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获异常并输出错误信息
return None # 请求失败返回None
def get_response_headers(self, url, headers=None, timeout=None):
"""
获取响应头
"""
try:
response = self.session.get(url, headers=headers, timeout=timeout) # 发送GET请求并获取响应
return response.headers # 返回响应头
except (Timeout, ConnectionError) as e:
print(f'Request failed: {e}') # 捕获异常并输出错误信息
return None # 请求失败返回None
def set_basic_auth(self, url, username, password):
"""
设置基本认证
"""
self.session.auth = HTTPBasicAuth(username, password) # 设置基本认证，传递用户名和密码
def disable_ssl_certificate_verification(self):
"""
禁用SSL证书验证
"""
self.session.verify = False # 禁用SSL证书验证，适用于开发环境或不安全的请求