126 lines
3.2 KiB
Python
126 lines
3.2 KiB
Python
from typing import List, Optional
|
|
from uuid import UUID
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from models.ship import Ship
|
|
from schemas.ship import ShipCreate, ShipUpdate
|
|
|
|
def get_all_ships(db: Session) -> List[Ship]:
|
|
"""
|
|
Retrieves all ships from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[Ship]: A list of all ship objects.
|
|
"""
|
|
return db.query(Ship).all()
|
|
|
|
def get_ship_by_id(db: Session, ship_id: UUID) -> Optional[Ship]:
|
|
"""
|
|
Retrieves a single ship by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
ship_id (UUID): The ID of the ship to retrieve.
|
|
|
|
Returns:
|
|
Optional[Ship]: The ship object if found, otherwise None.
|
|
"""
|
|
return db.query(Ship).filter(Ship.id == ship_id).first()
|
|
|
|
def get_ship_names(db: Session) -> List[str]:
|
|
"""
|
|
Retrieves a list of all ship names.
|
|
|
|
Note: This function is deprecated. Use get_all_ships() instead to get full ship objects.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[str]: A list of ship names.
|
|
"""
|
|
return [ship.name for ship in db.query(Ship.name).all()]
|
|
|
|
def create_ship(db: Session, ship_data: ShipCreate) -> Ship:
|
|
"""
|
|
Creates a new ship in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
ship_data (ShipCreate): The data for the ship to create.
|
|
|
|
Returns:
|
|
Ship: The newly created ship object.
|
|
|
|
Raises:
|
|
SQLAlchemyError: If there's an error creating the ship.
|
|
"""
|
|
try:
|
|
db_ship = Ship(**ship_data.dict())
|
|
db.add(db_ship)
|
|
db.commit()
|
|
db.refresh(db_ship)
|
|
return db_ship
|
|
except SQLAlchemyError:
|
|
db.rollback()
|
|
raise
|
|
|
|
def update_ship(db: Session, ship_id: UUID, ship_data: ShipUpdate) -> Optional[Ship]:
|
|
"""
|
|
Updates an existing ship in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
ship_id (UUID): The ID of the ship to update.
|
|
ship_data (ShipUpdate): The update data for the ship.
|
|
|
|
Returns:
|
|
Optional[Ship]: The updated ship object if found, otherwise None.
|
|
|
|
Raises:
|
|
SQLAlchemyError: If there's an error updating the ship.
|
|
"""
|
|
try:
|
|
db_ship = get_ship_by_id(db, ship_id)
|
|
if not db_ship:
|
|
return None
|
|
|
|
update_data = ship_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_ship, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_ship)
|
|
return db_ship
|
|
except SQLAlchemyError:
|
|
db.rollback()
|
|
raise
|
|
|
|
def delete_ship(db: Session, ship_id: UUID) -> bool:
|
|
"""
|
|
Deletes a ship from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
ship_id (UUID): The ID of the ship to delete.
|
|
|
|
Returns:
|
|
bool: True if the ship was deleted, False if not found.
|
|
|
|
Raises:
|
|
SQLAlchemyError: If there's an error deleting the ship.
|
|
"""
|
|
try:
|
|
db_ship = get_ship_by_id(db, ship_id)
|
|
if not db_ship:
|
|
return False
|
|
|
|
db.delete(db_ship)
|
|
db.commit()
|
|
return True
|
|
except SQLAlchemyError:
|
|
db.rollback()
|
|
raise |