57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import requests
|
|
|
|
|
|
class NotificationAPIClient:
|
|
def __init__(self, environment="dev"):
|
|
# 定义不同环境的域名
|
|
self.environment_urls = {
|
|
"dev": "http://dev.example.com",
|
|
"test": "http://192.168.1.202:9100/downhole-tool-system",
|
|
"prod": "http://prod.example.com"
|
|
}
|
|
|
|
# 设置当前环境的域名
|
|
if environment not in self.environment_urls:
|
|
raise ValueError(f"Invalid environment '{environment}'. Choose from: {list(self.environment_urls.keys())}")
|
|
|
|
self.base_url = self.environment_urls[environment]
|
|
|
|
def get_my_task(self, token):
|
|
# 拼接请求的完整 URL
|
|
url = f"{self.base_url}/notification/v1/my-task"
|
|
|
|
# 设置请求头，包含 Authorization Token
|
|
headers = {
|
|
"Authorization": f"Bearer {token}"
|
|
}
|
|
|
|
try:
|
|
# 发起 GET 请求
|
|
response = requests.get(url, headers=headers)
|
|
|
|
# 检查 HTTP 状态码并返回结果
|
|
if response.status_code == 200:
|
|
return response.json() # 返回解析后的 JSON 数据
|
|
else:
|
|
return {"error": f"HTTP {response.status_code}", "details": response.text}
|
|
except requests.RequestException as e:
|
|
return {"error": "Request failed", "details": str(e)}
|
|
|
|
|
|
# 使用示例
|
|
if __name__ == "__main__":
|
|
# 切换环境：可选 "dev", "test", "prod"
|
|
environment = "test"
|
|
|
|
# 创建客户端
|
|
client = NotificationAPIClient(environment=environment)
|
|
|
|
# 设置 Token
|
|
token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0ZW5hbnRfcm9sZV9rZXkiOiJzaHV6aXpob25neGluIiwidXNlcl9pZCI6IjEiLCJ1c2VyX25hbWUiOiJhZG1pbiIsInNjb3BlIjpbInNlcnZlciJdLCJleHAiOjE3MzY5Mjc5MTQsImp0aSI6InZQTm5iTUdiczJ6OHdPYi1VVTljcHVxd0oxVSIsImNsaWVudF9pZCI6IjFhN2JmY2M2MDI3NzRkNDk5NTkzNzU1MTFmYmIzYWYzIn0.5Z0yXfoLvJc4qEPf6RMS8Xw6VtjaNMPAyfQxEp0DIkM"
|
|
|
|
# 发起请求
|
|
result = client.get_my_task(token)
|
|
|
|
# 输出结果
|
|
print(result)
|