manager.py - master - ring0/RestaurantApp

manager.py
117 строк5 KB
ring0
Init project
28 май 2026, 20:53
28 май 2026, 20:53e63548a
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from models import Table, Reservation
from datetime import datetime

class TableManager:
    """Менеджер для работы со столиками"""
    
    def get_tables(self, db: Session, date: str = None, time: str = None):
        """Получение столиков с фильтрацией по дате и времени"""
        query = db.query(Table)
        
        if date and time:
            # Находим занятые столики на указанную дату и время
            busy_tables = db.query(Reservation.table_id).filter(
                and_(
                    Reservation.reservation_date == date,
                    Reservation.start_time <= time,
                    Reservation.end_time > time,
                    Reservation.status == 'confirmed'
                )
            ).subquery()
            
            # Исключаем занятые столики
            query = query.filter(Table.id.notin_(busy_tables.select()))
        
        return query.all()
    
    def update_status(self, db: Session, table_id: int, status: str):
        """Обновление статуса столика"""
        table = db.query(Table).filter(Table.id == table_id).first()
        if not table:
            return {"error": "Table not found"}
        
        table.status = status
        db.commit()
        db.refresh(table)
        return table.to_dict()

class ReservationManager:
    """Менеджер для работы с бронированиями"""
    
    def create_reservation(self, db: Session, reservation_data):
        """Создание бронирования с проверкой конфликтов"""
        # Проверка существования столика
        table = db.query(Table).filter(Table.id == reservation_data.table_id).first()
        if not table:
            return {"error": "Table not found"}
        
        # Проверка вместимости столика
        if reservation_data.guests_count > table.capacity:
            return {"error": f"Table capacity ({table.capacity}) is less than guests count ({reservation_data.guests_count})"}
        
        # Проверка конфликтов времени
        conflicts = db.query(Reservation).filter(
            and_(
                Reservation.table_id == reservation_data.table_id,
                Reservation.reservation_date == reservation_data.reservation_date,
                Reservation.status == 'confirmed',
                or_(
                    and_(
                        Reservation.start_time <= reservation_data.start_time,
                        Reservation.end_time > reservation_data.start_time
                    ),
                    and_(
                        Reservation.start_time < reservation_data.end_time,
                        Reservation.end_time >= reservation_data.end_time
                    ),
                    and_(
                        Reservation.start_time >= reservation_data.start_time,
                        Reservation.end_time <= reservation_data.end_time
                    )
                )
            )
        ).all()
        
        if conflicts:
            return {"error": "Time slot is already booked"}
        
        # Создание бронирования
        reservation = Reservation(
            table_id=reservation_data.table_id,
            customer_name=reservation_data.customer_name,
            reservation_date=reservation_data.reservation_date,
            start_time=reservation_data.start_time,
            end_time=reservation_data.end_time,
            guests_count=reservation_data.guests_count
        )
        
        db.add(reservation)
        db.commit()
        db.refresh(reservation)
        
        return reservation.to_dict()
    
    def get_reservations(self, db: Session, date: str = None, table_id: int = None):
        """Получение бронирований с фильтрацией"""
        query = db.query(Reservation)
        
        if date:
            query = query.filter(Reservation.reservation_date == date)
        
        if table_id:
            query = query.filter(Reservation.table_id == table_id)
        
        return query.all()
    
    def update_status(self, db: Session, reservation_id: int, status: str):
        """Обновление статуса бронирования"""
        reservation = db.query(Reservation).filter(Reservation.id == reservation_id).first()
        if not reservation:
            return {"error": "Reservation not found"}
        
        reservation.status = status
        db.commit()
        db.refresh(reservation)
        return reservation.to_dict()