online-food-ordering-system-mw6mst/helpers/online_order_helpers.py

from typing import List, Optional
from uuid import UUID
from sqlalchemy.orm import Session
from sqlalchemy import or_, and_
from models.online_order import OnlineOrder, OrderItem
from schemas.online_order import OnlineOrderCreate, OnlineOrderUpdate, OrderItemCreate, OrderItemUpdate

def get_online_order_by_id(db: Session, order_id: UUID) -> Optional[OnlineOrder]:
    """
    Retrieves an online order by its ID.

    Args:
        db (Session): The database session.
        order_id (UUID): The ID of the online order to retrieve.

    Returns:
        Optional[OnlineOrder]: The online order object if found, otherwise None.
    """
    return db.query(OnlineOrder).filter(OnlineOrder.id == order_id).first()

def get_all_online_orders(db: Session) -> List[OnlineOrder]:
    """
    Retrieves all online orders from the database.

    Args:
        db (Session): The database session.

    Returns:
        List[OnlineOrder]: A list of all online order objects.
    """
    return db.query(OnlineOrder).all()

def create_online_order(db: Session, order_data: OnlineOrderCreate) -> OnlineOrder:
    """
    Creates a new online order in the database.

    Args:
        db (Session): The database session.
        order_data (OnlineOrderCreate): The data for the online order to create.

    Returns:
        OnlineOrder: The newly created online order object.
    """
    db_order = OnlineOrder(**order_data.dict())
    db.add(db_order)
    db.commit()
    db.refresh(db_order)
    return db_order

def update_online_order(db: Session, order_id: UUID, order_data: OnlineOrderUpdate) -> Optional[OnlineOrder]:
    """
    Updates an existing online order in the database.

    Args:
        db (Session): The database session.
        order_id (UUID): The ID of the online order to update.
        order_data (OnlineOrderUpdate): The updated data for the online order.

    Returns:
        Optional[OnlineOrder]: The updated online order object if found, otherwise None.
    """
    db_order = get_online_order_by_id(db, order_id)
    if not db_order:
        return None

    for field, value in order_data.dict(exclude_unset=True).items():
        setattr(db_order, field, value)

    db.commit()
    db.refresh(db_order)
    return db_order

def delete_online_order(db: Session, order_id: UUID) -> bool:
    """
    Deletes an online order from the database.

    Args:
        db (Session): The database session.
        order_id (UUID): The ID of the online order to delete.

    Returns:
        bool: True if the online order was successfully deleted, False otherwise.
    """
    db_order = get_online_order_by_id(db, order_id)
    if not db_order:
        return False

    db.delete(db_order)
    db.commit()
    return True

def create_order_item(db: Session, order_id: UUID, item_data: OrderItemCreate) -> OrderItem:
    """
    Creates a new order item for an online order.

    Args:
        db (Session): The database session.
        order_id (UUID): The ID of the online order to add the item to.
        item_data (OrderItemCreate): The data for the order item to create.

    Returns:
        OrderItem: The newly created order item object.
    """
    db_order = get_online_order_by_id(db, order_id)
    if not db_order:
        raise ValueError(f"Online order with ID {order_id} not found.")

    db_item = OrderItem(**item_data.dict(), order_id=order_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def update_order_item(db: Session, item_id: UUID, item_data: OrderItemUpdate) -> Optional[OrderItem]:
    """
    Updates an existing order item.

    Args:
        db (Session): The database session.
        item_id (UUID): The ID of the order item to update.
        item_data (OrderItemUpdate): The updated data for the order item.

    Returns:
        Optional[OrderItem]: The updated order item object if found, otherwise None.
    """
    db_item = db.query(OrderItem).filter(OrderItem.id == item_id).first()
    if not db_item:
        return None

    for field, value in item_data.dict(exclude_unset=True).items():
        setattr(db_item, field, value)

    db.commit()
    db.refresh(db_item)
    return db_item

def delete_order_item(db: Session, item_id: UUID) -> bool:
    """
    Deletes an order item from the database.

    Args:
        db (Session): The database session.
        item_id (UUID): The ID of the order item to delete.

    Returns:
        bool: True if the order item was successfully deleted, False otherwise.
    """
    db_item = db.query(OrderItem).filter(OrderItem.id == item_id).first()
    if not db_item:
        return False

    db.delete(db_item)
    db.commit()
    return True

def search_online_orders(db: Session, query: str) -> List[OnlineOrder]:
    """
    Searches for online orders based on a search query.

    Args:
        db (Session): The database session.
        query (str): The search query string.

    Returns:
        List[OnlineOrder]: A list of online order objects matching the search query.
    """
    search_pattern = f"%{query}%"
    return db.query(OnlineOrder).filter(
        or_(
            OnlineOrder.order_number.ilike(search_pattern),
            OnlineOrder.customer_name.ilike(search_pattern),
            OnlineOrder.customer_email.ilike(search_pattern),
            OnlineOrder.customer_phone.ilike(search_pattern),
            OnlineOrder.delivery_address.ilike(search_pattern),
            and_(
                OrderItem.product_name.ilike(search_pattern),
                OrderItem.order_id == OnlineOrder.id,
            ),
        )
    ).all()