parent
7858495782
commit
e413247ada
@ -0,0 +1,114 @@
|
||||
"""
|
||||
API客户端工具类
|
||||
用于与FastAPI后端通信
|
||||
"""
|
||||
import requests
|
||||
from typing import Optional, Dict, List
|
||||
import json
|
||||
|
||||
class APIClient:
|
||||
"""API客户端"""
|
||||
|
||||
def __init__(self, base_url: str = "http://127.0.0.1:8000"):
|
||||
self.base_url = base_url.rstrip('/')
|
||||
|
||||
def _get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
|
||||
"""GET请求"""
|
||||
url = f"{self.base_url}{endpoint}"
|
||||
try:
|
||||
response = requests.get(url, params=params, timeout=5)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Exception(f"API请求失败: {str(e)}")
|
||||
|
||||
def _post(self, endpoint: str, data: Optional[Dict] = None, files: Optional[Dict] = None) -> Dict:
|
||||
"""POST请求"""
|
||||
url = f"{self.base_url}{endpoint}"
|
||||
try:
|
||||
if files:
|
||||
response = requests.post(url, files=files, timeout=30)
|
||||
else:
|
||||
response = requests.post(url, json=data, timeout=5)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Exception(f"API请求失败: {str(e)}")
|
||||
|
||||
def _delete(self, endpoint: str) -> Dict:
|
||||
"""DELETE请求"""
|
||||
url = f"{self.base_url}{endpoint}"
|
||||
try:
|
||||
response = requests.delete(url, timeout=5)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Exception(f"API请求失败: {str(e)}")
|
||||
|
||||
# 学生相关API
|
||||
def get_students(self) -> List[Dict]:
|
||||
"""获取学生列表"""
|
||||
return self._get("/api/students/")
|
||||
|
||||
def get_student(self, student_id: int) -> Dict:
|
||||
"""获取单个学生"""
|
||||
return self._get(f"/api/students/{student_id}")
|
||||
|
||||
def create_student(self, student_data: Dict) -> Dict:
|
||||
"""创建学生"""
|
||||
return self._post("/api/students/", data=student_data)
|
||||
|
||||
def import_students_from_excel(self, file_path: str) -> Dict:
|
||||
"""从Excel导入学生"""
|
||||
with open(file_path, 'rb') as f:
|
||||
files = {'file': (file_path.split('/')[-1], f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')}
|
||||
return self._post("/api/students/import-excel", files=files)
|
||||
|
||||
def delete_student(self, student_id: int) -> Dict:
|
||||
"""删除学生"""
|
||||
return self._delete(f"/api/students/{student_id}")
|
||||
|
||||
# 点名相关API
|
||||
def start_rollcall(self, rollcall_type: str, student_id: Optional[int] = None) -> Dict:
|
||||
"""开始点名"""
|
||||
data = {"rollcall_type": rollcall_type}
|
||||
if student_id:
|
||||
data["student_id"] = student_id
|
||||
return self._post("/api/rollcall/start", data=data)
|
||||
|
||||
def record_rollcall(self, record_data: Dict) -> Dict:
|
||||
"""记录点名结果"""
|
||||
return self._post("/api/rollcall/record", data=record_data)
|
||||
|
||||
def get_rollcall_records(self, skip: int = 0, limit: int = 100) -> List[Dict]:
|
||||
"""获取点名记录"""
|
||||
return self._get("/api/rollcall/records", params={"skip": skip, "limit": limit})
|
||||
|
||||
def transfer_rollcall(self, from_student_id: int, to_student_id: int) -> Dict:
|
||||
"""转移点名"""
|
||||
return self._post("/api/rollcall/transfer", data={
|
||||
"from_student_id": from_student_id,
|
||||
"to_student_id": to_student_id
|
||||
})
|
||||
|
||||
# 积分相关API
|
||||
def get_ranking(self, top_n: int = 10) -> Dict:
|
||||
"""获取积分排名"""
|
||||
return self._get("/api/scores/ranking", params={"top_n": top_n})
|
||||
|
||||
def export_scores(self, save_path: str) -> bool:
|
||||
"""导出积分详单"""
|
||||
url = f"{self.base_url}/api/scores/export"
|
||||
try:
|
||||
response = requests.get(url, timeout=30)
|
||||
response.raise_for_status()
|
||||
with open(save_path, 'wb') as f:
|
||||
f.write(response.content)
|
||||
return True
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Exception(f"导出失败: {str(e)}")
|
||||
|
||||
def get_statistics(self) -> Dict:
|
||||
"""获取统计信息"""
|
||||
return self._get("/api/scores/statistics")
|
||||
|
||||
Loading…
Reference in new issue