Implement basic aircraft management

api/database/aircraft.py

@@ -0,0 +1,87 @@
+from bson import ObjectId
+from fastapi import HTTPException
+
+from database.db import aircraft_collection
+from database.utils import aircraft_display_helper, aircraft_add_helper
+from schemas.aircraft import AircraftDisplaySchema, AircraftCreateSchema
+
+
+async def retrieve_aircraft(user: str = "") -> list[AircraftDisplaySchema]:
+    """
+    Retrieve a list of aircraft, optionally filtered by user
+
+    :param user: User to filter aircraft by
+    :return: List of aircraft
+    """
+    aircraft = []
+    if user == "":
+        async for doc in aircraft_collection.find():
+            aircraft.append(AircraftDisplaySchema(**aircraft_display_helper(doc)))
+    else:
+        async for doc in aircraft_collection.find({"user": ObjectId(user)}):
+            aircraft.append(AircraftDisplaySchema(**aircraft_display_helper(doc)))
+
+    return aircraft
+
+
+async def retrieve_aircraft_by_id(id: str) -> AircraftDisplaySchema:
+    """
+    Retrieve details about the requested aircraft
+
+    :param id: ID of desired aircraft
+    :return: Aircraft details
+    """
+    aircraft = await aircraft_collection.find_one({"_id": ObjectId(id)})
+
+    if aircraft is None:
+        raise HTTPException(404, "Aircraft not found")
+
+    return AircraftDisplaySchema(**aircraft_display_helper(aircraft))
+
+
+async def insert_aircraft(body: AircraftCreateSchema, id: str) -> ObjectId:
+    """
+    Insert a new aircraft into the database
+
+    :param body: Aircraft data
+    :param id: ID of creating user
+    :return: ID of inserted aircraft
+    """
+    aircraft = await aircraft_collection.insert_one(aircraft_add_helper(body.model_dump(), id))
+    return aircraft.inserted_id
+
+
+async def update_aircraft(body: AircraftCreateSchema, id: str) -> AircraftDisplaySchema:
+    """
+    Update given aircraft in the database
+
+    :param body: Updated aircraft data
+    :param id: ID of aircraft to update
+    :return: ID of updated aircraft
+    """
+    aircraft = await aircraft_collection.find_one({"_id": ObjectId(id)})
+
+    if aircraft is None:
+        raise HTTPException(404, "Aircraft not found")
+
+    updated_aircraft = await aircraft_collection.update_one({"_id": ObjectId(id)}, {"$set": body.model_dump()})
+    if updated_aircraft is None:
+        raise HTTPException(500, "Failed to update flight")
+
+    return id
+
+
+async def delete_aircraft(id: str) -> AircraftDisplaySchema:
+    """
+    Delete the given aircraft from the database
+
+    :param id: ID of aircraft to delete
+    :return: Deleted aircraft information
+    """
+    aircraft = await aircraft_collection.find_one({"_id": ObjectId(id)})
+
+    if aircraft is None:
+        raise HTTPException(404, "Aircraft not found")
+
+    await aircraft_collection.delete_one({"_id": ObjectId(id)})
+    return AircraftDisplaySchema(**aircraft_display_helper(aircraft))

api/database/db.py

@@ -24,4 +24,5 @@ except Exception as e:
 # Get db collections
 user_collection = db_client["user"]
 flight_collection = db_client["flight"]
+aircraft_collection = db_client["aircraft"]
 token_collection = db_client["token_blacklist"]

api/database/flights.py

@@ -88,8 +88,7 @@ async def retrieve_flight(id: str) -> FlightDisplaySchema:
     :param id: ID of flight to retrieve
     :return: Flight information
     """
-    oid = ObjectId(id)
-    flight = await flight_collection.find_one({"_id": oid})
+    flight = await flight_collection.find_one({"_id": ObjectId(id)})
 
     if flight is None:
         raise HTTPException(404, "Flight not found")

api/database/utils.py

@@ -3,6 +3,7 @@ import logging
 from bson import ObjectId
 
 from app.config import get_settings
+from schemas.aircraft import AircraftCategory, AircraftClass
 from .db import user_collection
 from routes.utils import get_hashed_password
 from schemas.user import AuthLevel, UserCreateSchema
@@ -13,6 +14,7 @@ logger = logging.getLogger("api")
 def user_helper(user) -> dict:
     """
     Convert given db response into a format usable by UserDisplaySchema
+
     :param user: Database response
     :return: Usable dict
     """
@@ -26,6 +28,7 @@ def user_helper(user) -> dict:
 def system_user_helper(user) -> dict:
     """
     Convert given db response to a format usable by UserSystemSchema
+
     :param user: Database response
     :return: Usable dict
     """
@@ -40,6 +43,7 @@ def system_user_helper(user) -> dict:
 def create_user_helper(user) -> dict:
     """
     Convert given db response to a format usable by UserCreateSchema
+
     :param user: Database response
     :return: Usable dict
     """
@@ -53,6 +57,7 @@ def create_user_helper(user) -> dict:
 def flight_display_helper(flight: dict) -> dict:
     """
     Convert given db response to a format usable by FlightDisplaySchema
+
     :param flight: Database response
     :return: Usable dict
     """
@@ -65,6 +70,7 @@ def flight_display_helper(flight: dict) -> dict:
 def flight_add_helper(flight: dict, user: str) -> dict:
     """
     Convert given flight schema and user string to a format that can be inserted into the db
+
     :param flight: Flight request body
     :param user: User that created flight
     :return: Combined dict that can be inserted into db
@@ -73,6 +79,36 @@ def flight_add_helper(flight: dict, user: str) -> dict:
     return flight
 
 
+def aircraft_add_helper(aircraft: dict, user: str) -> dict:
+    """
+    Convert given aircraft dict to a format that can be inserted into the db
+
+    :param aircraft: Aircraft request body
+    :param user: User that created aircraft
+    :return: Combined dict that can be inserted into db
+    """
+    aircraft["user"] = ObjectId(user)
+    aircraft["aircraft_category"] = aircraft["aircraft_category"].name
+    aircraft["aircraft_class"] = aircraft["aircraft_class"].name
+    return aircraft
+
+
+def aircraft_display_helper(aircraft: dict) -> dict:
+    """
+    Convert given db response into a format usable by AircraftDisplaySchema
+
+    :param aircraft:
+    :return: USable dict
+    """
+    aircraft["id"] = str(aircraft["_id"])
+    aircraft["user"] = str(aircraft["user"])
+    if aircraft["aircraft_category"] is not AircraftCategory:
+        aircraft["aircraft_category"] = AircraftCategory.__members__.get(aircraft["aircraft_category"])
+    if aircraft["aircraft_class"] is not AircraftClass:
+        aircraft["aircraft_class"] = AircraftClass.__members__.get(aircraft["aircraft_class"])
+    return aircraft
+
+
 # UTILS #
 
 async def create_admin_user():