224 lines
8.4 KiB
Python
224 lines
8.4 KiB
Python
import logging
|
|
from datetime import datetime
|
|
from typing import Any, List
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Form, UploadFile, File
|
|
|
|
from app.deps import get_current_user, admin_required
|
|
from database import flights as db
|
|
from database.flights import update_flight_fields
|
|
from database.img import upload_image
|
|
|
|
from schemas.flight import FlightConciseSchema, FlightDisplaySchema, FlightCreateSchema, FlightByDateSchema, \
|
|
FlightSchema
|
|
from schemas.user import UserDisplaySchema, AuthLevel
|
|
|
|
router = APIRouter()
|
|
|
|
logger = logging.getLogger("flights")
|
|
|
|
|
|
@router.get('/', summary="Get flights logged by the currently logged-in user", status_code=200)
|
|
async def get_flights(user: UserDisplaySchema = Depends(get_current_user), sort: str = "date", order: int = -1,
|
|
filter: str = "", filter_val: str = "") -> list[
|
|
FlightConciseSchema]:
|
|
"""
|
|
Get a list of the flights logged by the currently logged-in user
|
|
|
|
:param user: Current user
|
|
:param sort: Attribute to sort results by
|
|
:param order: Order of sorting (asc/desc)
|
|
:param filter: Field to filter results by
|
|
:param filter_val: Value to filter field by
|
|
:return: List of flights
|
|
"""
|
|
flights = await db.retrieve_flights(user.id, sort, order, filter, filter_val)
|
|
return flights
|
|
|
|
|
|
@router.get('/by-date', summary="Get flights logged by the current user, categorized by date", status_code=200,
|
|
response_model=dict)
|
|
async def get_flights_by_date(user: UserDisplaySchema = Depends(get_current_user), sort: str = "date",
|
|
order: int = -1, filter: str = "", filter_val: str = "") -> dict:
|
|
"""
|
|
Get a list of the flights logged by the currently logged-in user, categorized by year, month, and day
|
|
|
|
:param user: Current user
|
|
:param sort: Attribute to sort results by
|
|
:param order: Order of sorting (asc/desc)
|
|
:param filter: Field to filter results by
|
|
:param filter_val: Value to filter field by
|
|
:return:
|
|
"""
|
|
flights = await db.retrieve_flights(user.id, sort, order, filter, filter_val)
|
|
flights_ordered: FlightByDateSchema = {}
|
|
|
|
for flight in flights:
|
|
date = flight.date
|
|
flights_ordered.setdefault(date.year, {}).setdefault(date.month, {}).setdefault(date.day, []).append(flight)
|
|
|
|
return flights_ordered
|
|
|
|
|
|
@router.get('/totals', summary="Get total statistics for the current user", status_code=200, response_model=dict)
|
|
async def get_flight_totals(user: UserDisplaySchema = Depends(get_current_user), start_date: str = "",
|
|
end_date: str = "") -> dict:
|
|
"""
|
|
Get the total statistics for the currently logged-in user
|
|
|
|
:param user: Current user
|
|
:param start_date: Only count statistics after this date (optional)
|
|
:param end_date: Only count statistics before this date (optional)
|
|
:return: Dict of totals
|
|
"""
|
|
try:
|
|
start = datetime.strptime(start_date, "%Y-%m-%d") if start_date != "" else None
|
|
end = datetime.strptime(end_date, "%Y-%m-%d") if end_date != "" else None
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "Date range not processable")
|
|
|
|
return await db.retrieve_totals(user.id, start, end)
|
|
|
|
|
|
@router.get('/all', summary="Get all flights logged by all users", status_code=200,
|
|
dependencies=[Depends(admin_required)], response_model=list[FlightConciseSchema])
|
|
async def get_all_flights(sort: str = "date", order: int = -1) -> list[FlightConciseSchema]:
|
|
"""
|
|
Get a list of all flights logged by any user
|
|
|
|
:param sort: Attribute to sort results by
|
|
:param order: Order of sorting (asc/desc)
|
|
:return: List of flights
|
|
"""
|
|
flights = await db.retrieve_flights(sort=sort, order=order)
|
|
return flights
|
|
|
|
|
|
@router.get('/{flight_id}', summary="Get details of a given flight", response_model=FlightDisplaySchema,
|
|
status_code=200)
|
|
async def get_flight(flight_id: str, user: UserDisplaySchema = Depends(get_current_user)) -> FlightDisplaySchema:
|
|
"""
|
|
Get all details of a given flight
|
|
|
|
:param flight_id: ID of requested flight
|
|
:param user: Currently logged-in user
|
|
:return: Flight details
|
|
"""
|
|
flight = await db.retrieve_flight(flight_id)
|
|
if str(flight.user) != user.id and AuthLevel(user.level) != AuthLevel.ADMIN:
|
|
logger.info("Attempted access to unauthorized flight by %s", user.username)
|
|
raise HTTPException(403, "Unauthorized access")
|
|
|
|
return flight
|
|
|
|
|
|
@router.post('/', summary="Add a flight logbook entry", status_code=200)
|
|
async def add_flight(flight_body: FlightSchema, user: UserDisplaySchema = Depends(get_current_user)) -> dict:
|
|
"""
|
|
Add a flight logbook entry
|
|
|
|
:param flight_body: Information associated with new flight
|
|
:param images: Images associated with the new flight log
|
|
:param user: Currently logged-in user
|
|
:return: ID of newly created log
|
|
"""
|
|
|
|
flight_create = FlightCreateSchema(**flight_body.model_dump(), images=[])
|
|
|
|
flight = await db.insert_flight(flight_create, user.id)
|
|
|
|
return {"id": str(flight)}
|
|
|
|
|
|
@router.post('/{log_id}/add_images', summary="Add images to a flight log")
|
|
async def add_images(log_id: str, images: List[UploadFile] = File(...),
|
|
user: UserDisplaySchema = Depends(get_current_user)):
|
|
"""
|
|
Add images to a flight logbook entry
|
|
|
|
:param log_id: ID of flight log to add images to
|
|
:param images: Images to add
|
|
:param user: Currently logged-in user
|
|
:return: ID of updated flight
|
|
"""
|
|
flight = await db.retrieve_flight(log_id)
|
|
|
|
if not str(flight.user) == user.id and not user.level == AuthLevel.ADMIN:
|
|
raise HTTPException(403, "Unauthorized access")
|
|
|
|
image_ids = flight.images
|
|
|
|
if images:
|
|
for image in images:
|
|
image_response = await upload_image(image, user.id)
|
|
image_ids.append(image_response["file_id"])
|
|
|
|
return await update_flight_fields(log_id, dict(images=image_ids))
|
|
|
|
|
|
@router.put('/{flight_id}', summary="Update the given flight with new information", status_code=200)
|
|
async def update_flight(flight_id: str, flight_body: FlightCreateSchema,
|
|
user: UserDisplaySchema = Depends(get_current_user)) -> dict:
|
|
"""
|
|
Update the given flight with new information
|
|
|
|
:param flight_id: ID of flight to update
|
|
:param flight_body: New flight information to update with
|
|
:param user: Currently logged-in user
|
|
:return: ID of updated flight
|
|
"""
|
|
flight = await get_flight(flight_id, user)
|
|
if flight is None:
|
|
raise HTTPException(404, "Flight not found")
|
|
|
|
if str(flight.user) != user.id and AuthLevel(user.level) != AuthLevel.ADMIN:
|
|
logger.info("Attempted access to unauthorized flight by %s", user.username)
|
|
raise HTTPException(403, "Unauthorized access")
|
|
|
|
updated_flight_id = await db.update_flight(flight_body, flight_id)
|
|
|
|
return {"id": str(updated_flight_id)}
|
|
|
|
|
|
@router.patch('/{flight_id}', summary="Update a single field of the given flight with new information", status_code=200)
|
|
async def patch_flight(flight_id: str, update: dict,
|
|
user: UserDisplaySchema = Depends(get_current_user)) -> dict:
|
|
"""
|
|
Update a single field of the given flight
|
|
|
|
:param flight_id: ID of flight to update
|
|
:param update: Dictionary of fields and values to update
|
|
:param user: Currently logged-in user
|
|
:return: ID of updated flight
|
|
"""
|
|
flight = await get_flight(flight_id, user)
|
|
if flight is None:
|
|
raise HTTPException(404, "Flight not found")
|
|
|
|
if str(flight.user) != user.id and AuthLevel(user.level) != AuthLevel.ADMIN:
|
|
logger.info("Attempted access to unauthorized flight by %s", user.username)
|
|
raise HTTPException(403, "Unauthorized access")
|
|
|
|
updated_flight_id = await db.update_flight_fields(flight_id, update)
|
|
return {"id": str(updated_flight_id)}
|
|
|
|
|
|
@router.delete('/{flight_id}', summary="Delete the given flight", status_code=200, response_model=FlightDisplaySchema)
|
|
async def delete_flight(flight_id: str, user: UserDisplaySchema = Depends(get_current_user)) -> FlightDisplaySchema:
|
|
"""
|
|
Delete the given flight
|
|
|
|
:param flight_id: ID of flight to delete
|
|
:param user: Currently logged-in user
|
|
:return: 200
|
|
"""
|
|
flight = await get_flight(flight_id, user)
|
|
|
|
if str(flight.user) != user.id and AuthLevel(user.level) != AuthLevel.ADMIN:
|
|
logger.info("Attempted access to unauthorized flight by %s", user.username)
|
|
raise HTTPException(403, "Unauthorized access")
|
|
|
|
deleted = await db.delete_flight(flight_id)
|
|
|
|
return deleted
|