Flights API

main.py

from fastapi import FastAPI, Query, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from datetime import datetime, timedelta
from enum import Enum
import httpx
from typing import Optional, List
from pydantic import BaseModel
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Define response model
class FlightInfo(BaseModel):
    destination: str
    time: str
    company: str
    fnumber: str
    comment: str

# Define date options enum
class DateOption(str, Enum):
    yesterday = "yesterday"
    today = "today"
    tomorrow = "tomorrow"

# Define movement type enum
class MovementType(str, Enum):
    departures = "D"
    arrivals = "A"

# Define airport enum
class Airport(str, Enum):
    tunis = "tunis"
    djerba = "djerba"
    # Add more airports as needed
    
app = FastAPI(title="Tunisia Flights API", description="API to fetch flight information for Tunisian airports.")

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

def get_date_from_option(date_option: Optional[str]) -> datetime:
    current_date = datetime.now()
    if date_option == DateOption.yesterday:
        return current_date - timedelta(days=1)
    elif date_option == DateOption.tomorrow:
        return current_date + timedelta(days=1)
    return current_date

def validate_date(year: str, month: str, day: str) -> bool:
    try:
        datetime(year=int(year), month=int(month), day=int(day))
        return True
    except ValueError:
        return False

@app.get("/flights/", response_model=List[FlightInfo], summary="Get flight information", description="Fetch flight information for Tunisian airports based on date, airport, and movement type.")
async def get_flights(
    date_option: Optional[DateOption] = Query(None, description="Date option: yesterday, today, or tomorrow"),
    d: Optional[str] = Query(None, description="Day"),
    m: Optional[str] = Query(None, description="Month"),
    y: Optional[str] = Query(None, description="Year"),
    port: Airport = Query(Airport.tunis, description="Airport (tunis or djerba)"),
    dest: MovementType = Query(MovementType.departures, description="Movement type: D for departures, A for arrivals")
):
    """
    Get flight information for Tunisian airports.
    
    - **date_option**: Choose between yesterday, today, or tomorrow
    - **d**, **m**, **y**: Specific date (day, month, year)
    - **port**: Airport (tunis or djerba)
    - **dest**: Movement type (D for departures, A for arrivals)
    """
    # Determine the date to use
    if date_option:
        selected_date = get_date_from_option(date_option)
        day = selected_date.strftime("%d")
        month = selected_date.strftime("%m")
        year = selected_date.strftime("%Y")
    else:
        current_date = datetime.now()
        day = d if d else current_date.strftime("%d")
        month = m if m else current_date.strftime("%m")
        year = y if y else current_date.strftime("%Y")
    
    # Validate the date
    if not validate_date(year, month, day):
        raise HTTPException(status_code=400, detail="Invalid date provided.")
    
    # Construct the URL
    url = (
        f"https://www.oaca.nat.tn/vols/api/flight/filter"
        f"?frmmvtCod={dest}"
        f"&frmaeropVil=-1"
        f"&frmnumVol="
        f"&frmairport={port}"
        f"&frmday={day}"
        f"&frmmonth={month}"
        f"&frmacty={year}"
        f"&frmhour=0"
    )
    
    # Make the request using httpx
    async with httpx.AsyncClient(verify=False) as client:
        try:
            response = await client.get(url)
            response.raise_for_status()
            data = response.json()
            
            # Transform the data
            output_array = [
                FlightInfo(
                    destination=item["direction"],
                    time=item["heure"],
                    company=item["compagnie"].strip(),
                    fnumber=item["numVol"],
                    comment=item["commentaire"]
                )
                for item in data
            ]
            
            return output_array
            
        except httpx.RequestError as e:
            logger.error(f"Error fetching data: {str(e)}")
            raise HTTPException(status_code=503, detail=f"Error fetching data: {str(e)}")
        except Exception as e:
            logger.error(f"Internal server error: {str(e)}")
            raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)