116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
from typing import List, Optional, Union
|
|
import uuid
|
|
from sqlalchemy.orm import Session
|
|
from models.weather_forecast import WeatherForecast
|
|
from schemas.weather_forecast import WeatherForecastCreate, WeatherForecastUpdate, WeatherForecastSchema
|
|
from datetime import datetime
|
|
from sqlalchemy import func
|
|
|
|
def get_weather_forecasts(db: Session) -> List[WeatherForecastSchema]:
|
|
"""
|
|
Retrieves weather forecasts from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[WeatherForecastSchema]: A list of weather forecast objects.
|
|
"""
|
|
forecasts = db.query(WeatherForecast).all()
|
|
return [WeatherForecastSchema.from_orm(forecast) for forecast in forecasts]
|
|
|
|
def get_weather_forecast_by_id(db: Session, forecast_id: Union[str, uuid.UUID]) -> Optional[WeatherForecast]:
|
|
"""
|
|
Retrieves a weather forecast by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
forecast_id (Union[str, uuid.UUID]): The ID of the weather forecast.
|
|
|
|
Returns:
|
|
Optional[WeatherForecast]: The weather forecast object if found, otherwise None.
|
|
"""
|
|
return db.query(WeatherForecast).filter(WeatherForecast.id == forecast_id).first()
|
|
|
|
def create_weather_forecast(db: Session, forecast_data: WeatherForecastCreate) -> WeatherForecastSchema:
|
|
"""
|
|
Creates a new weather forecast in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
forecast_data (WeatherForecastCreate): The data for the weather forecast to create.
|
|
|
|
Returns:
|
|
WeatherForecastSchema: The newly created weather forecast object.
|
|
"""
|
|
db_forecast = WeatherForecast(**forecast_data.dict())
|
|
db.add(db_forecast)
|
|
db.commit()
|
|
db.refresh(db_forecast)
|
|
return WeatherForecastSchema.from_orm(db_forecast)
|
|
|
|
def update_weather_forecast(
|
|
db: Session, forecast_id: Union[str, uuid.UUID], forecast_data: WeatherForecastUpdate
|
|
) -> WeatherForecastSchema:
|
|
"""
|
|
Updates an existing weather forecast in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
forecast_id (Union[str, uuid.UUID]): The ID of the weather forecast to update.
|
|
forecast_data (WeatherForecastUpdate): The updated data for the weather forecast.
|
|
|
|
Returns:
|
|
WeatherForecastSchema: The updated weather forecast object.
|
|
"""
|
|
db_forecast = db.query(WeatherForecast).filter(WeatherForecast.id == forecast_id).first()
|
|
if not db_forecast:
|
|
raise ValueError(f"Weather forecast with ID {forecast_id} not found")
|
|
|
|
update_data = forecast_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_forecast, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_forecast)
|
|
return WeatherForecastSchema.from_orm(db_forecast)
|
|
|
|
def delete_weather_forecast(db: Session, forecast_id: Union[str, uuid.UUID]) -> bool:
|
|
"""
|
|
Deletes a weather forecast from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
forecast_id (Union[str, uuid.UUID]): The ID of the weather forecast to delete.
|
|
|
|
Returns:
|
|
bool: True if the weather forecast was successfully deleted, False otherwise.
|
|
"""
|
|
db_forecast = db.query(WeatherForecast).filter(WeatherForecast.id == forecast_id).first()
|
|
if not db_forecast:
|
|
return False
|
|
|
|
db.delete(db_forecast)
|
|
db.commit()
|
|
return True
|
|
|
|
def get_average_temperature_by_location(db: Session, location: str, start_date: datetime, end_date: datetime) -> Optional[float]:
|
|
"""
|
|
Retrieves the average temperature for a given location and date range.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
location (str): The location to filter by.
|
|
start_date (datetime): The start date of the date range (inclusive).
|
|
end_date (datetime): The end date of the date range (inclusive).
|
|
|
|
Returns:
|
|
Optional[float]: The average temperature if forecasts are found, otherwise None.
|
|
"""
|
|
avg_temp = db.query(func.avg(WeatherForecast.temperature)).filter(
|
|
WeatherForecast.location == location,
|
|
WeatherForecast.date >= start_date,
|
|
WeatherForecast.date <= end_date,
|
|
).scalar()
|
|
|
|
return avg_temp if avg_temp is not None else None |