```python import requests import json from typing import Dict, Any, Optional # ==================== 1. GET请求示例 ==================== def send_get_request(url: str, params: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: int = 10) -> Dict[str, Any]: """ 发送GET请求并处理响应 [ref_1] 参数: url: 请求的URL地址 params: 查询参数字典 headers: 请求头字典 timeout: 超时时间(秒) 返回: 包含响应信息的字典 """ result = {"success": False, "data": None, "error": None, "status_code": None} try: # 发送GET请求 [ref_1] response = requests.get( url=url, params=params, # 查询参数 [ref_1] headers=headers, # 请求头 [ref_1] timeout=timeout # 超时设置 [ref_1] ) # 记录状态码 result["status_code"] = response.status_code # 检查HTTP状态码 [ref_1] response.raise_for_status() # 如果状态码不是2xx，抛出HTTPError异常 [ref_1] # 根据Content-Type处理响应内容 [ref_4] content_type = response.headers.get('Content-Type', '') if 'application/json' in content_type: result["data"] = response.json() # JSON响应 [ref_4] else: # 设置正确的编码 [ref_1] response.encoding = response.apparent_encoding or 'utf-8' result["data"] = response.text # 文本响应 [ref_1] result["success"] = True except requests.exceptions.HTTPError as http_err: # HTTP错误处理 (4xx, 5xx) [ref_1] result["error"] = f"HTTP错误: {http_err}" result["data"] = response.text if 'response' in locals() else None except requests.exceptions.Timeout: # 超时处理 [ref_1] result["error"] = f"请求超时 ({timeout}秒)" except requests.exceptions.ConnectionError: # 连接错误处理 [ref_1] result["error"] = "连接服务器失败" except requests.exceptions.RequestException as req_err: # 其他请求异常 [ref_1] result["error"] = f"请求异常: {req_err}" except json.JSONDecodeError: # JSON解析错误 result["error"] = "JSON解析失败" result["data"] = response.text return result # ==================== 2. POST请求示例 ==================== def send_post_request(url: str, data_type: str = 'form', # 'form', 'json', 'file' data: Optional[Dict] = None, json_data: Optional[Dict] = None, files: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: int = 10) -> Dict[str, Any]: """ 发送POST请求并处理响应 [ref_1, ref_2, ref_5] 参数: url: 请求的URL地址 data_type: 数据类型 ('form', 'json', 'file') data: 表单数据 json_data: JSON数据 files: 文件数据 headers: 请求头 timeout: 超时时间 返回: 包含响应信息的字典 """ result = {"success": False, "data": None, "error": None, "status_code": None} # 设置默认请求头 default_headers = {'User-Agent': 'Mozilla/5.0'} if headers: default_headers.update(headers) try: if data_type == 'form': # 发送表单数据 [ref_1, ref_5] response = requests.post( url=url, data=data, # 表单数据 [ref_1] headers=default_headers, timeout=timeout ) elif data_type == 'json': # 发送JSON数据 [ref_1, ref_2] if not json_data and data: json_data = data # 确保Content-Type正确 [ref_2] default_headers['Content-Type'] = 'application/json' response = requests.post( url=url, json=json_data, # 自动序列化为JSON并设置Content-Type [ref_2] headers=default_headers, timeout=timeout ) elif data_type == 'file': # 文件上传 [ref_1] response = requests.post( url=url, files=files, # 文件数据 [ref_1] headers=default_headers, timeout=timeout ) else: result["error"] = f"不支持的数据类型: {data_type}" return result # 记录状态码 result["status_code"] = response.status_code # 状态码分类处理 status_code = response.status_code if 200 <= status_code < 300: # 2xx 成功 [ref_1] result["success"] = True content_type = response.headers.get('Content-Type', '') if 'application/json' in content_type: result["data"] = response.json() else: response.encoding = response.apparent_encoding or 'utf-8' result["data"] = response.text elif 300 <= status_code < 400: # 3xx 重定向 [ref_1] result["error"] = f"重定向状态码: {status_code}" result["redirect_url"] = response.headers.get('Location') elif 400 <= status_code < 500: # 4xx 客户端错误 [ref_1] result["error"] = f"客户端错误 ({status_code}): {response.reason}" result["data"] = response.text # 常见客户端错误处理 if status_code == 401: result["error"] += " - 需要身份验证" elif status_code == 403: result["error"] += " - 禁止访问" elif status_code == 404: result["error"] += " - 资源未找到" elif status_code == 429: result["error"] += " - 请求过于频繁" elif 500 <= status_code < 600: # 5xx 服务器错误 [ref_1] result["error"] = f"服务器错误 ({status_code}): {response.reason}" result["data"] = response.text else: result["error"] = f"未知状态码: {status_code}" except requests.exceptions.Timeout: result["error"] = f"请求超时 ({timeout}秒)" except requests.exceptions.ConnectionError: result["error"] = "连接服务器失败" except requests.exceptions.RequestException as req_err: result["error"] = f"请求异常: {req_err}" except json.JSONDecodeError: result["error"] = "JSON解析失败" if 'response' in locals(): result["data"] = response.text return result # ==================== 3. 状态码处理工具函数 ==================== def handle_status_code(response: requests.Response) -> Dict[str, Any]: """ 专门处理响应状态码的工具函数 [ref_1, ref_4] 参数: response: requests.Response对象 返回: 状态码处理结果字典 """ result = { "status_code": response.status_code, "category": "", "description": "", "should_retry": False, "action": "" } status_code = response.status_code # 状态码分类 [ref_1] if 100 <= status_code < 200: result["category"] = "信息响应" result["description"] = "请求已接收，继续处理" result["action"] = "继续等待或处理" elif 200 <= status_code < 300: result["category"] = "成功响应" result["description"] = "请求已成功处理" result["action"] = "处理响应数据" elif 300 <= status_code < 400: result["category"] = "重定向" result["description"] = "需要进一步操作以完成请求" result["action"] = "检查Location头并重定向" elif 400 <= status_code < 500: result["category"] = "客户端错误" result["description"] = "请求包含语法错误或无法完成" result["should_retry"] = False # 具体客户端错误处理 [ref_1] client_errors = { 400: ("错误请求", "检查请求参数和格式"), 401: ("未授权", "添加身份验证信息"), 403: ("禁止访问", "检查权限设置"), 404: ("未找到", "检查URL是否正确"), 429: ("请求过多", "降低请求频率") } if status_code in client_errors: desc, action = client_errors[status_code] result["description"] = desc result["action"] = action elif 500 <= status_code < 600: result["category"] = "服务器错误" result["description"] = "服务器在处理请求时发生错误" result["should_retry"] = True # 服务器错误通常可以重试 result["action"] = "稍后重试或联系服务器管理员" else: result["category"] = "未知状态码" result["description"] = "无法识别的状态码" return result # ==================== 4. 使用会话的高级示例 ==================== class RequestClient: """使用会话的请求客户端 [ref_1]""" def __init__(self, base_url: str = ""): self.session = requests.Session() self.base_url = base_url self.default_headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Accept': 'application/json', 'Accept-Encoding': 'gzip, deflate' } self.session.headers.update(self.default_headers) def get(self, endpoint: str, **kwargs) -> Dict[str, Any]: """发送GET请求""" url = f"{self.base_url}{endpoint}" if self.base_url else endpoint return send_get_request(url, **kwargs) def post(self, endpoint: str, data_type: str = 'json', **kwargs) -> Dict[str, Any]: """发送POST请求""" url = f"{self.base_url}{endpoint}" if self.base_url else endpoint return send_post_request(url, data_type=data_type, **kwargs) def set_auth(self, auth_type: str, **credentials): """设置认证信息""" if auth_type == 'basic': self.session.auth = (credentials.get('username'), credentials.get('password')) elif auth_type == 'bearer': self.session.headers['Authorization'] = f"Bearer {credentials.get('token')}" def close(self): """关闭会话""" self.session.close() # ==================== 5. 完整使用示例 ==================== def main(): """主函数演示完整用法""" print("=" * 60) print("Python Requests库 GET/POST请求演示") print("=" * 60) # 示例1: GET请求 - 获取公共API数据 print("\n1. GET请求示例 - 获取JSONPlaceholder数据") print("-" * 40) get_result = send_get_request( url="https://jsonplaceholder.typicode.com/posts/1", params={"_limit": "5"}, # 查询参数 [ref_1] headers={"Accept": "application/json"} # 请求头 [ref_1] ) print(f"状态码: {get_result['status_code']}") print(f"成功: {get_result['success']}") if get_result['success']: print(f"响应数据: {get_result['data']}") else: print(f"错误: {get_result['error']}") # 示例2: POST请求 - 提交表单数据 print("\n2. POST请求示例 - 提交表单数据") print("-" * 40) form_data = { "username": "test_user", "password": "test_pass123", "email": "test@example.com" } post_form_result = send_post_request( url="https://httpbin.org/post", data_type="form", data=form_data, # 表单数据 [ref_1] headers={"X-Custom-Header": "MyValue"} ) print(f"状态码: {post_form_result['status_code']}") print(f"成功: {post_form_result['success']}") # 示例3: POST请求 - 提交JSON数据 print("\n3. POST请求示例 - 提交JSON数据") print("-" * 40) json_payload = { "title": "测试文章", "body": "这是一篇测试文章的内容", "userId": 1 } post_json_result = send_post_request( url="https://jsonplaceholder.typicode.com/posts", data_type="json", json_data=json_payload # JSON数据 [ref_2] ) print(f"状态码: {post_json_result['status_code']}") print(f"成功: {post_json_result['success']}") if post_json_result['success']: print(f"创建的资源ID: {post_json_result['data'].get('id', 'N/A')}") # 示例4: 状态码详细处理 print("\n4. 状态码处理演示") print("-" * 40) # 模拟各种状态码响应 test_responses = [ ("成功响应", 200), ("未找到", 404), ("服务器错误", 500), ("重定向", 301) ] for desc, code in test_responses: # 创建模拟响应对象 mock_response = requests.Response() mock_response.status_code = code mock_response.reason = "Test Reason" status_info = handle_status_code(mock_response) print(f"{desc} ({code}):") print(f" 类别: {status_info['category']}") print(f" 描述: {status_info['description']}") print(f" 建议操作: {status_info['action']}") print(f" 是否重试: {status_info['should_retry']}") print() # 示例5: 使用会话客户端 print("\n5. 使用会话客户端演示") print("-" * 40) client = RequestClient(base_url="https://jsonplaceholder.typicode.com") # 设置认证信息 client.set_auth('bearer', token="your_token_here") # 使用会话发送多个请求 posts_result = client.get("/posts") if posts_result['success']: print(f"获取到 {len(posts_result['data'])} 篇文章") # 创建新文章 new_post = { "title": "会话测试文章", "body": "使用会话对象创建的测试文章", "userId": 1 } create_result = client.post("/posts", data_type='json', json_data=new_post) if create_result['success']: print(f"成功创建文章，ID: {create_result['data'].get('id')}") client.close() print("\n" + "=" * 60) print("演示完成") print("=" * 60) # ==================== 6. 最佳实践总结表格 ==================== """ | 场景 | 方法 | 关键参数 | 状态码处理 | 异常处理 | 参考来源 | |------|------|----------|------------|----------|----------| | **获取数据** | `requests.get()` | `params`, `headers`, `timeout` | 检查2xx状态码，使用`raise_for_status()` | 捕获`HTTPError`, `Timeout`, `ConnectionError` | [ref_1] | | **提交表单** | `requests.post()` | `data`, `headers` | 检查201/200状态码 | 同上 | [ref_1], [ref_5] | | **提交JSON** | `requests.post()` | `json`, `headers` | 检查201状态码 | 同上 | [ref_2], [ref_6] | | **文件上传** | `requests.post()` | `files`, `headers` | 检查200状态码 | 同上 | [ref_1] | | **会话管理** | `requests.Session()` | 跨请求保持cookies/headers | 每次请求单独检查 | 统一异常处理 | [ref_1] | | **错误处理** | `try-except`块 | - | 分类处理4xx/5xx | 使用`requests.exceptions` | [ref_1] | """ if __name__ == "__main__": main() ``` ## **核心要点说明** ### **1. GET请求的关键步骤** - **基础请求**: 使用`requests.get(url)`发送最简单的GET请求 [ref_1] - **参数传递**: 通过`params`参数传递查询字符串，Requests会自动编码 [ref_1] - **请求头设置**: 使用`headers`参数设置User-Agent等头部信息 [ref_1] - **超时控制**: 使用`timeout`参数防止请求无限期挂起 [ref_1] ### **2. POST请求的三种数据格式** | 数据类型 | 参数 | Content-Type | 适用场景 | 参考 | |----------|------|--------------|----------|------| | **表单数据** | `data` | application/x-www-form-urlencoded | 传统表单提交、登录 | [ref_1], [ref_5] | | **JSON数据** | `json` | application/json | RESTful API、现代Web服务 | [ref_2], [ref_6]