Spaces:
Paused
Paused
| from typing import List, Optional | |
| from fastapi import HTTPException, status, Depends | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from app.models.orders import Order, OrderItem, OrderStatus | |
| from ..db.crud import ( | |
| create_order_db, | |
| get_order_by_id, | |
| update_order_status, | |
| get_orders_by_status, | |
| get_orders_by_date_range | |
| ) | |
| from ..db.database import get_db | |
| class OrderService: | |
| def __init__(self, db: AsyncSession = Depends(get_db)): | |
| self.db = db | |
| async def create_order(self, order: Order) -> Order: | |
| """Create a new order.""" | |
| if not order.items: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Order must contain items" | |
| ) | |
| return await create_order_db(self.db, order) | |
| async def get_order(self, order_id: int) -> Optional[Order]: | |
| """Get order by ID.""" | |
| order = await get_order_by_id(self.db, order_id) | |
| if not order: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail="Order not found" | |
| ) | |
| return order | |
| async def update_status(self, order_id: int, new_status: OrderStatus) -> Order: | |
| """Update order status.""" | |
| order = await get_order_by_id(self.db, order_id) | |
| if not order: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail="Order not found" | |
| ) | |
| return await update_order_status(self.db, order_id, new_status) | |
| async def get_orders_by_status(self, status: OrderStatus) -> List[Order]: | |
| """Get orders by status.""" | |
| return await get_orders_by_status(self.db, status) | |
| async def get_daily_orders(self, date: str) -> List[Order]: | |
| """Get orders for a specific date.""" | |
| return await get_orders_by_date_range(self.db, date, date) | |