weather_api/main.py

import utils
from typing import Optional
from pydantic import BaseModel
from fastapi import FastAPI, status, HTTPException
from db_connector import database,engine,DB_REBUILD


if DB_REBUILD == 'True':
    database.metadata.drop_all(engine)
    utils.initialize_database()


class Location(BaseModel):
    id: int = None,
    name: str = None,
    country: str = None,
    longitude: float = None,
    latitude: float = None,
    user: int = None,
    
class Users(BaseModel):
    id: int = None,
    name: str = None,
    email: str = None,


class Config(BaseModel):
    id: int = None,
    user_id: int = None,
    location_id: int = None,



app = FastAPI()

def error_4xx_handler(return_data: dict) -> None:
    """
    Handles errors with status code 400.

    Checks if the key 'error' exists in the `return_data` dictionary.
    If it does, it raises an HTTPException with status code 400 and the value
    of the 'error' key as the detail.

    Args:
        return_data (dict): Data returned from an API endpoint.

    Raises:
        HTTPException: If the 'error' key is present in `return_data`.

    """

    # Check if 'error' key exists in the return_data dictionary
    if 'error' in return_data.keys():
        # Raise HTTPException with status code 400 and the 'error' value as detail
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={'error': return_data['error']}
        )


@app.get("/")
async def index_response():
    """
    A function that returns a status of "OK" when the root URL is accessed.
    No parameters are passed, and it returns a dictionary with the status.
    """
    return {"Status": "OK"}


@app.get("/locations")
async def get_location_weather(
    places: Optional[str] = 'all',
    user: Optional[int] = None
    ):
    """
    A function to retrieve weather data for specified locations.
    
    Parameters:
    places (str): A string containing location identifiers separated by commas.

    Returns:
    dict: A dictionary containing weather data for the specified locations.
    """    
    #NOTE: Add option for fetching weather data for all locations, (debugging purposes)

    if places == 'database_locations':
        result = utils.get_database_locations(user=None)
    elif places == 'configured':
        result = utils.get_database_locations(user)
    else:
        if places != 'all':
            places = [int(x) for x in list(places.split(","))]
        result = utils.retrieve_weather_data(places)
        error_4xx_handler(result)
        
    return result


@app.get("/locations/{id}")
async def get_weather_by_id(id: int):
    """
    Retrieves weather data for a specified location ID.

    Parameters:
        id (int): The ID of the location to retrieve weather data for.

    Returns:
        dict: A dictionary containing the retrieved weather data for the specified location.
    """
    
    result = utils.retrieve_weather_data([id])
    error_4xx_handler(result)
    return result
    

@app.post("/locations", status_code=status.HTTP_201_CREATED)
async def add_location(loc: Location):
    """
    Add a new location to the database.

    Parameters:
    - loc (Location): The location object to be added.

    Raises:
    - HTTPException: If the name exceeds 200 characters.

    Returns:
    None
    """

    if len(loc.name.encode('utf-8')) > 200:
        raise HTTPException(
            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail={'name': 'Name cannot be longer than 200 characters'}
        )
    else:
        id = int(loc.id)
        #if loc.name!= 'existing':
        #    id = utils.get_available_ids(1)[0]
        utils.add_location({"id":id ,"name":loc.name,"country":loc.country,"longitude":loc.longitude,"latitude":loc.latitude, "user": loc.user}, no_commit=False)


@app.delete("/locations/{id}")
async def delete_location(id: int):
    """
    Delete a location by its ID.

    Parameters:
    id (int): The ID of the location to be deleted.
    Returns:
    None
    """

    print(f"deleted location {id}")
    utils.delete_location(id)



@app.get("/location/search")
async def search_location(query: str):
    """
    A function to retrieve search results based on the provided query string.
    
    Parameters:
    query (str): The search query string.
    
    Returns:
    The search results based on the provided query.
    """
    result = utils.search_location(query)
    return result

@app.post("/location/disable")
async def search_location(id: int,
                          user: int
):

    result = utils.config_disable_location(id, user)
    return result