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()