52 lines
No EOL
1.7 KiB
Python
52 lines
No EOL
1.7 KiB
Python
from typing import List, Optional
|
|
import uuid
|
|
from datetime import datetime
|
|
from models import Order, Drink
|
|
from websocket_manager import websocket_manager
|
|
from prometheus_metrics import drink_metrics
|
|
|
|
|
|
class OrderDatabase:
|
|
def __init__(self):
|
|
self.orders: dict[str, Order] = {}
|
|
|
|
async def create_order(self, drinks: List[Drink]) -> Order:
|
|
"""Create a new order with the given drinks."""
|
|
order_id = str(uuid.uuid4())
|
|
order = Order(
|
|
id=order_id,
|
|
order_date=datetime.now(),
|
|
drinks=drinks
|
|
)
|
|
self.orders[order_id] = order
|
|
|
|
# Zeichne verkaufte Getränke für Prometheus-Statistiken auf
|
|
drink_metrics.record_drinks(drinks)
|
|
|
|
# WebSocket-Broadcast für neue Bestellung
|
|
await websocket_manager.broadcast_order_created(order.model_dump())
|
|
|
|
return order
|
|
|
|
def get_all_orders(self) -> List[Order]:
|
|
"""Get all orders sorted by order date (oldest first)."""
|
|
return sorted(self.orders.values(), key=lambda order: order.order_date)
|
|
|
|
def get_order_by_id(self, order_id: str) -> Optional[Order]:
|
|
"""Get a specific order by ID."""
|
|
return self.orders.get(order_id)
|
|
|
|
async def delete_order(self, order_id: str) -> bool:
|
|
"""Delete an order by ID. Returns True if successful, False if not found."""
|
|
if order_id in self.orders:
|
|
del self.orders[order_id]
|
|
|
|
# WebSocket-Broadcast für gelöschte Bestellung
|
|
await websocket_manager.broadcast_order_deleted(order_id)
|
|
|
|
return True
|
|
return False
|
|
|
|
|
|
# Global database instance
|
|
db = OrderDatabase() |