50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
import httpx
|
|
from typing import Any, Dict
|
|
|
|
|
|
class WeatherAPIError(Exception):
|
|
"""天气API调用异常"""
|
|
|
|
def __init__(self, status_code: int, message: str):
|
|
self.status_code = status_code
|
|
self.message = message
|
|
super().__init__(f"Weather API error {status_code}: {message}")
|
|
|
|
|
|
class WeatherAPIClient:
|
|
"""简单的天气数据客户端"""
|
|
|
|
def __init__(self, base_url: str, api_key: str, timeout: float = 15.0):
|
|
self.base_url = base_url.rstrip('/') if base_url else ''
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
|
|
async def get_current_weather(self, city: str, *, units: str = 'metric', lang: str = 'zh_cn') -> Dict[str, Any]:
|
|
if not self.api_key:
|
|
raise WeatherAPIError(401, '天气服务未配置 API 密钥')
|
|
|
|
endpoint = f"{self.base_url}/weather" if self.base_url else "https://api.openweathermap.org/data/2.5/weather"
|
|
params = {
|
|
'q': city,
|
|
'appid': self.api_key,
|
|
'units': units,
|
|
'lang': lang,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(endpoint, params=params)
|
|
except httpx.RequestError as exc:
|
|
raise WeatherAPIError(0, str(exc)) from exc
|
|
|
|
if response.status_code != 200:
|
|
detail = ''
|
|
try:
|
|
payload = response.json()
|
|
detail = payload.get('message', '')
|
|
except ValueError:
|
|
detail = response.text
|
|
raise WeatherAPIError(response.status_code, detail or '天气服务调用失败')
|
|
|
|
return response.json()
|