143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
from requests import Session
|
|
from typing import Dict, Any, Optional
|
|
|
|
|
|
class APIRequest:
|
|
def __init__(self, base_url: str, auth_params: Dict[str, str]):
|
|
"""
|
|
初始化API请求工具类
|
|
|
|
Args:
|
|
base_url: API的基础URL
|
|
auth_params: 认证所需的参数，如用户名、密码等
|
|
"""
|
|
self.base_url = base_url
|
|
self.auth_params = auth_params
|
|
self.session = Session()
|
|
self.token = None
|
|
self._authenticate()
|
|
|
|
def _authenticate(self) -> None:
|
|
"""获取并设置访问令牌"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{self.base_url}/oauth/token", params=self.auth_params
|
|
)
|
|
response.raise_for_status() # 检查请求是否成功
|
|
self.token = response.json()["access_token"]
|
|
self.session.headers.update({"Authorization": f"Bearer {self.token}"})
|
|
print("认证成功，token已设置")
|
|
except Exception as e:
|
|
print(f"认证失败: {e}")
|
|
self.token = None
|
|
|
|
def request(self, method: str, endpoint: str, **kwargs) -> Any:
|
|
"""
|
|
发送API请求
|
|
|
|
Args:
|
|
method: HTTP方法，如'GET', 'POST', 'PUT', 'DELETE'等
|
|
endpoint: API端点，不包含基础URL部分
|
|
**kwargs: 其他requests.request支持的参数
|
|
|
|
Returns:
|
|
响应的JSON数据或原始文本
|
|
|
|
Raises:
|
|
Exception: 请求失败时抛出异常
|
|
"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
# 自动添加access_token参数（如果需要）
|
|
if self.token and "params" in kwargs and isinstance(kwargs["params"], dict):
|
|
kwargs["params"].setdefault("access_token", self.token)
|
|
|
|
try:
|
|
response = self.session.request(method, url, verify=False, **kwargs)
|
|
response.raise_for_status() # 检查请求是否成功
|
|
|
|
# 根据响应内容类型返回不同格式的数据
|
|
if response.headers.get("Content-Type", "").startswith("application/json"):
|
|
return response.json()
|
|
else:
|
|
return response.text
|
|
except Exception as e:
|
|
print(f"请求失败: {e}")
|
|
raise
|
|
|
|
def get(self, endpoint: str, **kwargs) -> Any:
|
|
"""发送GET请求"""
|
|
return self.request("GET", endpoint, **kwargs)
|
|
|
|
def post(self, endpoint: str, **kwargs) -> Any:
|
|
"""发送POST请求"""
|
|
return self.request("POST", endpoint, **kwargs)
|
|
|
|
def put(self, endpoint: str, **kwargs) -> Any:
|
|
"""发送PUT请求"""
|
|
return self.request("PUT", endpoint, **kwargs)
|
|
|
|
def delete(self, endpoint: str, **kwargs) -> Any:
|
|
"""发送DELETE请求"""
|
|
return self.request("DELETE", endpoint, **kwargs)
|
|
|
|
|
|
# 创建API请求实例的工厂函数
|
|
def create(
|
|
ip: str = "https://band.hxzhxy.cn", auth_params: Optional[Dict[str, str]] = None
|
|
):
|
|
"""
|
|
创建API请求实例
|
|
|
|
Args:
|
|
ip: API服务器的IP地址或域名
|
|
auth_params: 认证参数，默认为预定义的值
|
|
|
|
Returns:
|
|
APIRequest实例
|
|
"""
|
|
# 默认的认证参数
|
|
DEFAULT_AUTH_PARAMS = {
|
|
"username": "tangchao",
|
|
"password": "123456a",
|
|
"client_id": "client",
|
|
"grant_type": "password",
|
|
"client_secret": "123456",
|
|
}
|
|
|
|
# 使用用户提供的认证参数或默认值
|
|
auth_params = auth_params or DEFAULT_AUTH_PARAMS
|
|
|
|
# 构建基础URL
|
|
if not ip.startswith(("http://", "https://")):
|
|
base_url = f"http://{ip}"
|
|
else:
|
|
base_url = ip
|
|
|
|
return APIRequest(base_url, auth_params)
|
|
|
|
|
|
# 示例使用
|
|
if __name__ == "__main__":
|
|
# 通过IP地址创建API实例
|
|
# api = create_api_instance("192.168.0.244:8100")
|
|
|
|
# 发送GET请求
|
|
# response = api.get("/net-manager/band/exportExcelNew", params={"keyword": "", "type": 0, "teamId": 19})
|
|
# print(response)
|
|
|
|
# 发送POST请求
|
|
# response = api.post("/studentLeave/add", json={
|
|
# "leaveType": "事假",
|
|
# "leaveReason": "超长",
|
|
# "beginDate": "2025-07-24",
|
|
# "endDate": "2025-07-24",
|
|
# "beginSection": 0,
|
|
# "endSection": 0,
|
|
# "imgUrl": "[{\"url\":\"http://192.168.0.244:9000/test-iot-school/2025-07-24/3489fb7492a04cb08958134686a27293.png\"}]",
|
|
# "isCycle": 1,
|
|
# "stuId": "7957"
|
|
# })
|
|
# print(response)
|
|
pass
|