106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from app.api.deps import get_db, get_current_active_user
|
|
from app.models.ppr import Airport
|
|
from app.schemas.ppr import Airport as AirportSchema
|
|
from app.models.ppr import User
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/lookup/{code_or_name}", response_model=List[AirportSchema])
|
|
async def lookup_airport_by_code_or_name(
|
|
code_or_name: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""
|
|
Lookup airport by ICAO code or name.
|
|
If input is 4 characters and all uppercase letters, treat as ICAO code.
|
|
Otherwise, search by name.
|
|
"""
|
|
clean_input = code_or_name.strip().upper()
|
|
|
|
if len(clean_input) < 2:
|
|
return []
|
|
|
|
# Check if input looks like an ICAO code (4 letters)
|
|
if len(clean_input) == 4 and clean_input.isalpha():
|
|
# Exact ICAO match first
|
|
airport = db.query(Airport).filter(Airport.icao == clean_input).first()
|
|
if airport:
|
|
return [airport]
|
|
# Then search ICAO codes that start with input
|
|
airports = db.query(Airport).filter(
|
|
Airport.icao.like(clean_input + "%")
|
|
).limit(5).all()
|
|
return airports
|
|
else:
|
|
# Search by name (case-insensitive partial match)
|
|
airports = db.query(Airport).filter(
|
|
Airport.name.ilike("%" + code_or_name + "%")
|
|
).limit(10).all()
|
|
return airports
|
|
|
|
|
|
@router.get("/search", response_model=List[AirportSchema])
|
|
async def search_airports(
|
|
q: Optional[str] = Query(None, description="Search query for ICAO code, IATA code, or airport name"),
|
|
limit: int = Query(10, ge=1, le=100, description="Maximum number of results"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""
|
|
Search airports by ICAO code, IATA code, or name.
|
|
"""
|
|
if not q or len(q.strip()) < 2:
|
|
return []
|
|
|
|
search_term = q.strip()
|
|
clean_search = search_term.upper()
|
|
|
|
# Search across ICAO, IATA, and name fields
|
|
airports = db.query(Airport).filter(
|
|
(Airport.icao.like("%" + clean_search + "%")) |
|
|
(Airport.iata.like("%" + clean_search + "%")) |
|
|
(Airport.name.ilike("%" + search_term + "%")) |
|
|
(Airport.city.ilike("%" + search_term + "%"))
|
|
).limit(limit).all()
|
|
|
|
return airports
|
|
|
|
|
|
@router.get("/public/lookup/{code_or_name}", response_model=List[AirportSchema])
|
|
async def public_lookup_airport_by_code_or_name(
|
|
code_or_name: str,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Public lookup airport by ICAO code or name.
|
|
If input is 4 characters and all uppercase letters, treat as ICAO code.
|
|
Otherwise, search by name.
|
|
No authentication required.
|
|
"""
|
|
clean_input = code_or_name.strip().upper()
|
|
|
|
if len(clean_input) < 2:
|
|
return []
|
|
|
|
# Check if input looks like an ICAO code (4 letters)
|
|
if len(clean_input) == 4 and clean_input.isalpha():
|
|
# Exact ICAO match first
|
|
airport = db.query(Airport).filter(Airport.icao == clean_input).first()
|
|
if airport:
|
|
return [airport]
|
|
# Then search ICAO codes that start with input
|
|
airports = db.query(Airport).filter(
|
|
Airport.icao.like(clean_input + "%")
|
|
).limit(5).all()
|
|
return airports
|
|
else:
|
|
# Search by name (case-insensitive partial match)
|
|
airports = db.query(Airport).filter(
|
|
Airport.name.ilike("%" + code_or_name + "%")
|
|
).limit(10).all()
|
|
return airports |