Automated Action 1468af1391 Add Weather Data API with OpenWeatherMap integration
- Created FastAPI application with SQLite database integration
- Implemented OpenWeatherMap client with caching
- Added endpoints for current weather, forecasts, and history
- Included comprehensive error handling and validation
- Set up Alembic migrations
- Created detailed README with usage examples

generated with BackendIM... (backend.im)
2025-05-12 14:26:44 +00:00

143 lines
4.3 KiB
Python

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from typing import Union, Dict, Any
import logging
# Configure logger
logger = logging.getLogger(__name__)
class WeatherAPIException(HTTPException):
"""Base exception class for the Weather API."""
def __init__(
self,
status_code: int,
detail: Any = None,
headers: Dict[str, str] = None,
error_code: str = None
):
self.error_code = error_code
super().__init__(status_code=status_code, detail=detail, headers=headers)
class ExternalAPIError(WeatherAPIException):
"""Exception raised when external API calls fail."""
def __init__(
self,
detail: Any = "External API error",
status_code: int = 503,
headers: Dict[str, str] = None,
error_code: str = "external_api_error"
):
super().__init__(
status_code=status_code,
detail=detail,
headers=headers,
error_code=error_code
)
class InvalidParameterError(WeatherAPIException):
"""Exception raised when request parameters are invalid."""
def __init__(
self,
detail: Any = "Invalid parameters",
status_code: int = 400,
headers: Dict[str, str] = None,
error_code: str = "invalid_parameter"
):
super().__init__(
status_code=status_code,
detail=detail,
headers=headers,
error_code=error_code
)
class ResourceNotFoundError(WeatherAPIException):
"""Exception raised when a requested resource is not found."""
def __init__(
self,
detail: Any = "Resource not found",
status_code: int = 404,
headers: Dict[str, str] = None,
error_code: str = "resource_not_found"
):
super().__init__(
status_code=status_code,
detail=detail,
headers=headers,
error_code=error_code
)
# Error handlers
async def http_exception_handler(request: Request, exc: Union[HTTPException, StarletteHTTPException]) -> JSONResponse:
"""
Handle HTTPExceptions and return a standardized JSON response.
"""
headers = getattr(exc, "headers", None)
# Log the exception
logger.error(f"HTTP Exception: {exc.status_code} - {exc.detail}")
# Add more context if it's our custom exception
if isinstance(exc, WeatherAPIException):
content = {
"error": {
"code": exc.error_code,
"message": exc.detail
},
"status_code": exc.status_code
}
else:
content = {
"error": {
"code": "http_error",
"message": exc.detail
},
"status_code": exc.status_code
}
return JSONResponse(
status_code=exc.status_code,
content=content,
headers=headers
)
async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
"""
Handle validation errors from Pydantic models and return a standardized error response.
"""
# Log the exception
logger.error(f"Validation error: {exc.errors()}")
# Create a more user-friendly error detail
errors = []
for error in exc.errors():
loc = " -> ".join([str(loc_item) for loc_item in error["loc"]])
errors.append({
"location": loc,
"message": error["msg"],
"type": error["type"]
})
content = {
"error": {
"code": "validation_error",
"message": "Request validation error",
"details": errors
},
"status_code": 422
}
return JSONResponse(
status_code=422,
content=content
)
# Configure application-wide exception handlers
def configure_exception_handlers(app):
"""
Configure exception handlers for the FastAPI application.
"""
app.add_exception_handler(HTTPException, http_exception_handler)
app.add_exception_handler(StarletteHTTPException, http_exception_handler)
app.add_exception_handler(RequestValidationError, validation_exception_handler)