88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
from datetime import datetime, timezone
|
|
from os import environ
|
|
import json
|
|
|
|
import requests
|
|
|
|
URL = "https://api.weatherapi.com/v1"
|
|
REQUEST_TIMEOUT_SECONDS = 10
|
|
|
|
|
|
def time_now() -> str:
|
|
return str(datetime.now(timezone.utc))
|
|
|
|
|
|
def get_current_weather(location: str) -> str:
|
|
weather = Weather(location=location)
|
|
return json.dumps(weather.realtime())
|
|
|
|
|
|
def get_weather_forecast(location: str, days: int = 0) -> str:
|
|
if days is None:
|
|
days = 0
|
|
|
|
days = int(days) + 1
|
|
days = max(1, days)
|
|
days = min(14, days)
|
|
|
|
weather = Weather(location=location)
|
|
return json.dumps(weather.forecast(days=days))
|
|
|
|
|
|
class Weather:
|
|
def __init__(self, location: str):
|
|
self.__location = location
|
|
self.api_key = environ.get("WEATHER_API_KEY")
|
|
|
|
@property
|
|
def location(self) -> str:
|
|
return self.__location
|
|
|
|
@staticmethod
|
|
def make_request(method: str, params: dict) -> dict:
|
|
try:
|
|
response = requests.get(
|
|
url=f"{URL}{method}",
|
|
params=params,
|
|
timeout=REQUEST_TIMEOUT_SECONDS,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
except requests.RequestException as error:
|
|
return {"error": f"Weather API request failed: {error}"}
|
|
except ValueError as error:
|
|
return {"error": f"Weather API returned invalid JSON: {error}"}
|
|
|
|
if isinstance(payload, dict) and "error" in payload:
|
|
api_error = payload["error"]
|
|
if isinstance(api_error, dict):
|
|
message = api_error.get("message", "Unknown weather API error")
|
|
else:
|
|
message = str(api_error)
|
|
return {"error": message}
|
|
|
|
return payload
|
|
|
|
def realtime(self):
|
|
method = "/current.json"
|
|
params = {
|
|
"key": self.api_key,
|
|
"q": self.location,
|
|
}
|
|
return self.make_request(method=method, params=params)
|
|
|
|
def forecast(self, days: int = 14):
|
|
method = "/forecast.json"
|
|
params = {
|
|
"key": self.api_key,
|
|
"q": self.location,
|
|
"days": days,
|
|
}
|
|
return self.make_request(method=method, params=params)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_weather = Weather("Aqtobe")
|
|
result = json.dumps(test_weather.forecast(days=13), indent=2)
|
|
print(result)
|