DuckDB Installation

import duckdb
import pandas as pd
from typing import List, Dict, Any, Optional

class WorldDatabase:
    def __init__(self, db_path: str = 'world.duckdb'):
        self.db_path = db_path
        self.conn = duckdb.connect(db_path)
        
    def query(self, sql: str, params: List = None) -> List[Dict]:
        """Execute query and return results as list of dictionaries"""
        if params:
            result = self.conn.execute(sql, params)
        else:
            result = self.conn.execute(sql)
        
        columns = [desc[0] for desc in result.description]
        return [dict(zip(columns, row)) for row in result.fetchall()]
    
    def query_df(self, sql: str, params: List = None) -> pd.DataFrame:
        """Execute query and return results as pandas DataFrame"""
        if params:
            return self.conn.execute(sql, params).df()
        else:
            return self.conn.execute(sql).df()
    
    def get_countries(self, region: Optional[str] = None, limit: Optional[int] = None) -> pd.DataFrame:
        sql = "SELECT * FROM countries"
        params = []
        
        if region:
            sql += " WHERE region = ?"
            params.append(region)
            
        if limit:
            sql += f" LIMIT {limit}"
            
        return self.query_df(sql, params if params else None)
    
    def get_cities_by_country(self, country_name: str, limit: Optional[int] = None) -> pd.DataFrame:
        sql = "SELECT * FROM cities WHERE country_name = ?"
        params = [country_name]
        
        if limit:
            sql += f" LIMIT {limit}"
            
        return self.query_df(sql, params)
    
    def analyze_population_by_region(self) -> pd.DataFrame:
        """Analytical query example"""
        sql = """
        SELECT 
            region,
            COUNT(DISTINCT c.name) as countries,
            COUNT(DISTINCT s.name) as states,
            COUNT(DISTINCT cities.name) as cities,
            AVG(TRY_CAST(cities.latitude AS DOUBLE)) as avg_latitude,
            AVG(TRY_CAST(cities.longitude AS DOUBLE)) as avg_longitude
        FROM countries c
        LEFT JOIN states s ON c.id = s.country_id  
        LEFT JOIN cities ON c.id = cities.country_id
        WHERE c.region IS NOT NULL
        GROUP BY c.region
        ORDER BY cities DESC
        """
        return self.query_df(sql)
    
    def find_cities_near_coordinates(self, lat: float, lon: float, max_distance_km: float = 100) -> pd.DataFrame:
        """Find cities within a certain distance using Haversine formula"""
        sql = """
        SELECT 
            name,
            country_name,
            state_name,
            latitude,
            longitude,
            6371 * acos(
                cos(radians(?)) * cos(radians(TRY_CAST(latitude AS DOUBLE))) *
                cos(radians(TRY_CAST(longitude AS DOUBLE)) - radians(?)) +
                sin(radians(?)) * sin(radians(TRY_CAST(latitude AS DOUBLE)))
            ) as distance_km
        FROM cities
        WHERE latitude IS NOT NULL 
        AND longitude IS NOT NULL
        AND 6371 * acos(
            cos(radians(?)) * cos(radians(TRY_CAST(latitude AS DOUBLE))) *
            cos(radians(TRY_CAST(longitude AS DOUBLE)) - radians(?)) +
            sin(radians(?)) * sin(radians(TRY_CAST(latitude AS DOUBLE)))
        ) <= ?
        ORDER BY distance_km
        """
        return self.query_df(sql, [lat, lon, lat, lat, lon, lat, max_distance_km])
    
    def get_country_statistics(self) -> pd.DataFrame:
        """Get comprehensive country statistics"""
        sql = """
        SELECT 
            c.region,
            COUNT(DISTINCT c.name) as countries,
            COUNT(DISTINCT c.currency) as currencies,  
            COUNT(DISTINCT s.name) as states,
            COUNT(cities.name) as cities,
            string_agg(DISTINCT c.currency, ', ') as currency_list
        FROM countries c
        LEFT JOIN states s ON c.id = s.country_id
        LEFT JOIN cities ON c.id = cities.country_id  
        WHERE c.region IS NOT NULL
        GROUP BY c.region
        ORDER BY cities DESC
        """
        return self.query_df(sql)
    
    def close(self):
        self.conn.close()

# Usage example
db = WorldDatabase()

try:
    # Basic queries
    print("=== European Countries ===")
    europe_countries = db.get_countries(region="Europe", limit=5)
    print(europe_countries[['name', 'iso2', 'capital', 'currency']])
    
    print("\n=== US Cities ===")
    us_cities = db.get_cities_by_country("United States", limit=5)
    print(us_cities[['name', 'state_name', 'latitude', 'longitude']])
    
    # Analytical queries
    print("\n=== Population Analysis by Region ===")
    population_analysis = db.analyze_population_by_region()
    print(population_analysis)
    
    print("\n=== Cities near New York (within 100km) ===")
    nearby_cities = db.find_cities_near_coordinates(40.7128, -74.0060, 100)
    print(nearby_cities.head(10))
    
    print("\n=== Country Statistics ===")
    stats = db.get_country_statistics()
    print(stats)

finally:
    db.close()