from fastapi import FastAPI, Request, Query, HTTPException from fastapi.responses import JSONResponse from typing import Optional import uvicorn from models.weather_models import ( WeatherResponse, AutocompleteResponse, ErrorResponse, ErrorDetail, RequestModel, ) from services.api_client import lifespan app = FastAPI(title="Weatherstack-Compatible API", lifespan=lifespan) @app.get("/health", include_in_schema=False) async def health(): return {"status": "ok", "message": "Weatherhack API is live"} import os from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): weatherstack_access_key: str = Field("test", alias="WEATHERSTACK_ACCESS_KEY") model_config = SettingsConfigDict( env_file=".env", extra="ignore", ) settings = Settings() # Full debug for HF environment if "WEATHERSTACK_ACCESS_KEY" in os.environ: env_val = os.environ["WEATHERSTACK_ACCESS_KEY"] print(f"DEBUG: Found in os.environ. Length: {len(env_val)}") else: print("DEBUG: NOT found in os.environ") print( f"DEBUG: Settings value: {settings.weatherstack_access_key[:4]}...{settings.weatherstack_access_key[-4:]}" ) def validate_key(access_key: str): if access_key != settings.weatherstack_access_key: return ErrorResponse( success=False, error=ErrorDetail( code=101, type="unauthorized", info="Invalid API access key." ), ) return None from services.location_service import get_location_details from services.weather_service import fetch_current_weather @app.get("/current", response_model=WeatherResponse) async def get_current_weather( access_key: str, query: str, units: str = "m", language: str = "en", callback: Optional[str] = None, ): error = validate_key(access_key) if error: return JSONResponse(status_code=401, content=error.model_dump()) location = await get_location_details(query) if not location: return JSONResponse( status_code=404, content=ErrorResponse( success=False, error=ErrorDetail( code=602, type="invalid_query", info="Invalid location query." ), ).model_dump(), ) try: current_weather = await fetch_current_weather( float(location.lat), float(location.lon), unit=units ) response = WeatherResponse( request=RequestModel( type="City", # This should be dynamic based on query type query=query, language=language, unit=units, ), location=location, current=current_weather, ) return response except Exception as e: return JSONResponse( status_code=500, content=ErrorResponse( success=False, error=ErrorDetail(code=615, type="request_failed", info=str(e)), ).model_dump(), ) from services.weather_service import fetch_historical_weather, fetch_weather_forecast from services.location_service import get_autocomplete_suggestions @app.get("/historical", response_model=WeatherResponse) async def get_historical_weather( access_key: str, query: str, historical_date: Optional[str] = None, historical_date_start: Optional[str] = None, historical_date_end: Optional[str] = None, units: str = "m", language: str = "en", hourly: int = 0, interval: int = 3, ): error = validate_key(access_key) if error: return JSONResponse(status_code=401, content=error.model_dump()) location = await get_location_details(query) if not location: return JSONResponse( status_code=404, content=ErrorResponse( success=False, error=ErrorDetail( code=602, type="invalid_query", info="Invalid location query." ), ).model_dump(), ) date = historical_date or historical_date_start if not date: return JSONResponse( status_code=400, content=ErrorResponse( success=False, error=ErrorDetail( code=614, type="missing_historical_date", info="No date provided." ), ).model_dump(), ) try: hist_data = await fetch_historical_weather( float(location.lat), float(location.lon), date, unit=units ) return WeatherResponse( request=RequestModel( type="City", query=query, language=language, unit=units ), location=location, historical=hist_data, ) except Exception as e: return JSONResponse( status_code=500, content=ErrorResponse( success=False, error=ErrorDetail(code=615, type="request_failed", info=str(e)), ).model_dump(), ) @app.get("/forecast", response_model=WeatherResponse) async def get_weather_forecast( access_key: str, query: str, forecast_days: int, units: str = "m", language: str = "en", hourly: int = 0, interval: int = 3, ): error = validate_key(access_key) if error: return JSONResponse(status_code=401, content=error.model_dump()) location = await get_location_details(query) if not location: return JSONResponse( status_code=404, content=ErrorResponse( success=False, error=ErrorDetail( code=602, type="invalid_query", info="Invalid location query." ), ).model_dump(), ) try: forecast_data = await fetch_weather_forecast( float(location.lat), float(location.lon), forecast_days, unit=units ) return WeatherResponse( request=RequestModel( type="City", query=query, language=language, unit=units ), location=location, forecast=forecast_data, ) except Exception as e: return JSONResponse( status_code=500, content=ErrorResponse( success=False, error=ErrorDetail(code=615, type="request_failed", info=str(e)), ).model_dump(), ) @app.get("/autocomplete", response_model=AutocompleteResponse) async def autocomplete_location(access_key: str, query: str): error = validate_key(access_key) if error: return JSONResponse(status_code=401, content=error.model_dump()) suggestions = await get_autocomplete_suggestions(query) return AutocompleteResponse( request={"query": query, "type": "City"}, locations=suggestions ) from fastapi.responses import HTMLResponse, RedirectResponse @app.get("/", include_in_schema=False) async def root(): return RedirectResponse(url="/dashboard") @app.get("/dashboard", response_class=HTMLResponse, include_in_schema=False) async def dashboard(): return """
Premium, Zero-Cost Weather Infrastructure