与API密钥管理器服务交互的客户端。
| 4 | from typing import Optional, Union |
| 5 | |
| 6 | class APIKeyManagerClient: |
| 7 | """与API密钥管理器服务交互的客户端。""" |
| 8 | |
| 9 | def __init__(self, base_url: str = "http://localhost:8002"): |
| 10 | """初始化客户端。 |
| 11 | |
| 12 | 参数: |
| 13 | base_url: API密钥管理器服务的基础URL |
| 14 | """ |
| 15 | self.base_url = base_url.rstrip('/') |
| 16 | |
| 17 | def get_api_key(self, source_name: str) -> Optional[str]: |
| 18 | """获取特定源的API密钥。 |
| 19 | |
| 20 | 参数: |
| 21 | source_name: 接口供应商名称 |
| 22 | |
| 23 | 返回: |
| 24 | API密钥,如果请求失败则返回None |
| 25 | """ |
| 26 | try: |
| 27 | response = requests.post( |
| 28 | f"{self.base_url}/get_apikey", |
| 29 | json={"source_name": source_name}, |
| 30 | timeout=5 |
| 31 | ) |
| 32 | if response.status_code == 200: |
| 33 | return response.json().get("api_key") |
| 34 | else: |
| 35 | print(f"获取API密钥失败: {response.status_code} - {response.text}") |
| 36 | return None |
| 37 | except Exception as e: |
| 38 | print(f"获取API密钥时出错: {str(e)}") |
| 39 | return None |
| 40 | |
| 41 | def notice_api_key_usage( |
| 42 | self, |
| 43 | api_key: str, |
| 44 | model_name: str, |
| 45 | source_name: str, |
| 46 | create_time: Union[datetime, str], |
| 47 | finish_time: Union[datetime, str], |
| 48 | execution_time: float, |
| 49 | status: bool, |
| 50 | prompt_tokens: Optional[int] = None, |
| 51 | completion_tokens: Optional[int] = None, |
| 52 | request_id: Optional[str] = None, |
| 53 | remark: Optional[str] = "" |
| 54 | ) -> bool: |
| 55 | """通知服务关于API密钥的使用情况。 |
| 56 | |
| 57 | 参数: |
| 58 | api_key: 使用的API密钥 |
| 59 | model_name: 模型名称 |
| 60 | source_name: 接口供应商名称 |
| 61 | create_time: 请求开始时间 |
| 62 | finish_time: 请求完成时间 |
| 63 | execution_time: 执行时间(秒) |
no outgoing calls
no test coverage detected