HTTPS API客户端功能优化求助：适配认证与端点要求的脚本完善

核心优化点 #

• 统一认证头处理：替换原Authorization为要求的AUTH_TOKEN请求头
• 封装API操作：将POST/PUT/GET/DELETE方法封装为可复用的类方法
• 响应标准化解析：统一处理不同请求的响应格式，提取用户数据与错误信息
• 参数校验：对PUT/DELETE等必须传user_id的方法做参数约束
• 环境适配：支持通过变量配置域名与令牌，便于不同环境切换

完整优化代码 #

import requests
import json
from typing import Optional, Dict, List

class UserAPIClient:
    def __init__(self, auth_token: str, domain: str = "interview.canadacentral.cloudapp.azure.com"):
        self.auth_token = auth_token
        self.base_url = f"https://{domain}/v1/user"
        self.headers = {
            "AUTH_TOKEN": self.auth_token,
            "Content-Type": "application/json"
        }

    def _handle_response(self, response: requests.Response) -> Dict:
        """统一处理API响应，解析返回格式"""
        try:
            data = response.json()
        except json.JSONDecodeError:
            return {"users": [], "errors": [f"无效响应格式: {response.text}"]}
        
        # 处理错误状态码
        if response.status_code >= 400:
            error_msg = data.get("errors", [f"请求失败，状态码: {response.status_code}"])
            return {"users": [], "errors": error_msg}
        
        return data

    def create_user(self, username: str, password: str) -> Dict:
        """创建用户（POST请求）"""
        payload = {"username": username, "password": password}
        response = requests.post(self.base_url, headers=self.headers, json=payload)
        return self._handle_response(response)

    def update_user(self, user_id: int, username: str, password: str) -> Dict:
        """更新用户（PUT请求，必须指定user_id）"""
        params = {"user_id": user_id}
        payload = {"username": username, "password": password}
        response = requests.put(self.base_url, headers=self.headers, params=params, json=payload)
        return self._handle_response(response)

    def get_users(self, user_id: Optional[int] = None) -> Dict:
        """获取用户信息（GET请求，可指定user_id或返回全部）"""
        params = {"user_id": user_id} if user_id else None
        response = requests.get(self.base_url, headers=self.headers, params=params)
        return self._handle_response(response)

    def delete_user(self, user_id: int) -> Dict:
        """删除用户（DELETE请求，必须指定user_id）"""
        params = {"user_id": user_id}
        response = requests.delete(self.base_url, headers=self.headers, params=params)
        return self._handle_response(response)

# 示例使用
if __name__ == "__main__":
    # 配置令牌与域名（可从环境变量读取，适配不同环境）
    API_TOKEN = "YOUR_AUTH_TOKEN_HERE"
    DOMAIN = "interview.canadacentral.cloudapp.azure.com"

    client = UserAPIClient(API_TOKEN, DOMAIN)

    # 1. 创建用户
    create_result = client.create_user("test_user", "test_pass")
    if create_result["errors"]:
        print(f"创建用户失败: {create_result['errors']}")
    else:
        user_id = create_result["users"][0]["user_id"]
        print(f"创建用户成功，ID: {user_id}")

    # 2. 获取全部用户
    get_all_result = client.get_users()
    if get_all_result["errors"]:
        print(f"获取全部用户失败: {get_all_result['errors']}")
    else:
        print("全部用户信息:")
        for user in get_all_result["users"]:
            print(f"ID: {user['user_id']}, 用户名: {user['username']}")

    # 3. 更新用户（使用刚创建的user_id）
    if 'user_id' in locals():
        update_result = client.update_user(user_id, "updated_user", "updated_pass")
        if update_result["errors"]:
            print(f"更新用户失败: {update_result['errors']}")
        else:
            print("用户更新成功")

    # 4. 删除用户（使用刚创建的user_id）
    if 'user_id' in locals():
        delete_result = client.delete_user(user_id)
        if delete_result["errors"]:
            print(f"删除用户失败: {delete_result['errors']}")
        else:
            print("用户删除成功")

优化说明 #

• 类封装：将客户端逻辑封装为UserAPIClient类，便于复用与维护，初始化时传入认证令牌与域名，支持不同环境切换
• 认证头修正：严格按照API要求使用AUTH_TOKEN请求头，替换原代码中的错误认证头格式
• 响应处理：通过_handle_response统一解析响应，处理JSON解码错误、HTTP错误状态码，返回标准化的users与errors结构
• 参数约束：PUT/DELETE方法强制要求传入user_id，符合API规范
• 环境适配：支持通过变量配置令牌与域名，可轻松从环境变量读取（如os.getenv("AUTH_TOKEN")），适配开发/测试/生产环境
• 类型提示：使用Python类型提示增强代码可读性与IDE支持

内容的提问来源于stack exchange，提问作者Rahul