|
|
import os |
|
|
import sys |
|
|
|
|
|
src_directory=os.path.abspath(os.path.join(os.path.dirname(__file__),"../..","pydantic")) |
|
|
sys.path.append(src_directory) |
|
|
|
|
|
|
|
|
|
|
|
import pandas as pd |
|
|
from utils.logger import logger |
|
|
from fastapi import HTTPException |
|
|
from models.schemas import ContinentPopulationResponse, CountryPopulation |
|
|
|
|
|
file_path = "world_population.csv" |
|
|
|
|
|
|
|
|
try: |
|
|
countries_df = pd.read_csv(file_path) |
|
|
countries_df["Population"] = pd.to_numeric(countries_df["Population"], errors="coerce") |
|
|
countries_df["Area"] = pd.to_numeric(countries_df["Area"], errors="coerce") |
|
|
logger.info("CSV file loaded successfully.") |
|
|
except Exception as e: |
|
|
logger.error(f"Error loading CSV: {e}") |
|
|
raise RuntimeError("Failed to load CSV file.") |
|
|
|
|
|
def get_continent_population(continent_name: str) -> ContinentPopulationResponse: |
|
|
try: |
|
|
logger.info(f"Fetching data for continent: {continent_name}") |
|
|
continent_group = countries_df[countries_df["Continent"] == continent_name].dropna(subset=["Population", "Area"]) |
|
|
|
|
|
if continent_group.empty: |
|
|
raise HTTPException(status_code=404, detail="Continent not found.") |
|
|
|
|
|
total_population = int(continent_group["Population"].sum()) |
|
|
total_area = int(continent_group["Area"].sum()) |
|
|
if total_area == 0: |
|
|
raise HTTPException(status_code=400, detail="Total area is zero.") |
|
|
|
|
|
max_row = continent_group.loc[continent_group["Population"].idxmax()] |
|
|
min_row = continent_group.loc[continent_group["Population"].idxmin()] |
|
|
avg_population = int(continent_group["Population"].mean()) |
|
|
|
|
|
return ContinentPopulationResponse( |
|
|
continent=continent_name, |
|
|
total_population=total_population, |
|
|
total_area=total_area, |
|
|
continent_population_density=total_population / total_area, |
|
|
max_population=CountryPopulation( |
|
|
country=max_row["Country"], population=int(max_row["Population"]) |
|
|
), |
|
|
min_population=CountryPopulation( |
|
|
country=min_row["Country"], population=int(min_row["Population"]) |
|
|
), |
|
|
avg_population=avg_population |
|
|
) |
|
|
except HTTPException as http_err: |
|
|
logger.error(f"HTTP exception: {http_err.detail}") |
|
|
raise http_err |
|
|
except Exception as e: |
|
|
logger.error(f"Unexpected error: {e}") |
|
|
raise HTTPException(status_code=500, detail="Internal server error.") |
|
|
|
