91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from models.ride import Ride
|
|
from schemas.ride import RideCreate, RideSchema
|
|
|
|
def get_all_rides(db: Session) -> List[RideSchema]:
|
|
"""
|
|
Retrieve all dispatched rides from the database.
|
|
|
|
Args:
|
|
db (Session): SQLAlchemy database session.
|
|
|
|
Returns:
|
|
List[RideSchema]: List of ride schemas.
|
|
"""
|
|
rides = db.query(Ride).all()
|
|
return [RideSchema.from_orm(ride) for ride in rides]
|
|
|
|
def create_ride(db: Session, ride: RideCreate) -> Optional[RideSchema]:
|
|
"""
|
|
Create a new ride in the database.
|
|
|
|
Args:
|
|
db (Session): SQLAlchemy database session.
|
|
ride (RideCreate): Ride data for creation.
|
|
|
|
Returns:
|
|
Optional[RideSchema]: Ride schema if created successfully, None otherwise.
|
|
"""
|
|
db_ride = Ride(**ride.dict())
|
|
db.add(db_ride)
|
|
try:
|
|
db.commit()
|
|
db.refresh(db_ride)
|
|
return RideSchema.from_orm(db_ride)
|
|
except Exception as e:
|
|
db.rollback()
|
|
return None
|
|
|
|
def get_ride_by_id(db: Session, ride_id: int) -> Optional[RideSchema]:
|
|
"""
|
|
Retrieve a ride by its ID from the database.
|
|
|
|
Args:
|
|
db (Session): SQLAlchemy database session.
|
|
ride_id (int): ID of the ride to retrieve.
|
|
|
|
Returns:
|
|
Optional[RideSchema]: Ride schema if found, None otherwise.
|
|
"""
|
|
ride = db.query(Ride).filter(Ride.id == ride_id).first()
|
|
return RideSchema.from_orm(ride) if ride else None
|
|
|
|
def update_ride(db: Session, ride_id: int, ride_data: RideCreate) -> Optional[RideSchema]:
|
|
"""
|
|
Update an existing ride in the database.
|
|
|
|
Args:
|
|
db (Session): SQLAlchemy database session.
|
|
ride_id (int): ID of the ride to update.
|
|
ride_data (RideCreate): Updated ride data.
|
|
|
|
Returns:
|
|
Optional[RideSchema]: Updated ride schema if successful, None otherwise.
|
|
"""
|
|
ride = db.query(Ride).filter(Ride.id == ride_id).first()
|
|
if ride:
|
|
for key, value in ride_data.dict(exclude_unset=True).items():
|
|
setattr(ride, key, value)
|
|
db.commit()
|
|
db.refresh(ride)
|
|
return RideSchema.from_orm(ride)
|
|
return None
|
|
|
|
def delete_ride(db: Session, ride_id: int) -> bool:
|
|
"""
|
|
Delete a ride from the database.
|
|
|
|
Args:
|
|
db (Session): SQLAlchemy database session.
|
|
ride_id (int): ID of the ride to delete.
|
|
|
|
Returns:
|
|
bool: True if the ride was deleted successfully, False otherwise.
|
|
"""
|
|
ride = db.query(Ride).filter(Ride.id == ride_id).first()
|
|
if ride:
|
|
db.delete(ride)
|
|
db.commit()
|
|
return True
|
|
return False |