database.py - master - vacya/restaurant-booking

database.py
148 строк4 KB
Евгений
first_commit
28 май 2026, 19:28
28 май 2026, 19:2898d4533
"""Модуль для работы с базой данных."""

import sqlite3
from typing import List, Dict, Any

DATABASE = 'booking.db'


def get_db_connection():
    """Получить соединение с БД."""
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    """Инициализировать базу данных."""
    conn = get_db_connection()
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS tables (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            number INTEGER UNIQUE NOT NULL,
            seats INTEGER NOT NULL,
            location TEXT
        )
    ''')

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS bookings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_name TEXT NOT NULL,
            customer_phone TEXT NOT NULL,
            table_id INTEGER NOT NULL,
            date TEXT NOT NULL,
            time TEXT NOT NULL,
            guests INTEGER NOT NULL,
            status TEXT DEFAULT 'active',
            FOREIGN KEY (table_id) REFERENCES tables(id)
        )
    ''')

    cursor.execute("SELECT COUNT(*) FROM tables")
    if cursor.fetchone()[0] == 0:
        tables_data = [
            (1, 2, 'у окна'),
            (2, 2, 'у окна'),
            (3, 4, 'в центре'),
            (4, 4, 'в центре'),
            (5, 6, 'веранда'),
            (6, 8, 'вип-зона'),
        ]
        cursor.executemany(
            "INSERT INTO tables (number, seats, location) VALUES (?, ?, ?)",
            tables_data
        )

    conn.commit()
    conn.close()


def get_all_tables() -> List[Dict[str, Any]]:
    """Получить все столики."""
    conn = get_db_connection()
    tables = conn.execute("SELECT * FROM tables ORDER BY number").fetchall()
    conn.close()
    return [dict(row) for row in tables]


def get_available_tables(date: str, time: str, guests: int) -> List[Dict[str, Any]]:
    """Получить свободные столики."""
    conn = get_db_connection()
    query = '''
        SELECT t.* FROM tables t
        WHERE t.seats >= ?
        AND t.id NOT IN (
            SELECT table_id FROM bookings
            WHERE date = ? AND time = ? AND status = 'active'
        )
        ORDER BY t.seats, t.number
    '''
    tables = conn.execute(query, (guests, date, time)).fetchall()
    conn.close()
    return [dict(row) for row in tables]


def create_booking(name: str, phone: str, table_id: int,
                   date: str, time: str, guests: int) -> int:
    """Создать бронирование."""
    conn = get_db_connection()
    cursor = conn.execute('''
        INSERT INTO bookings (customer_name, customer_phone, table_id, date, time, guests)
        VALUES (?, ?, ?, ?, ?, ?)
    ''', (name, phone, table_id, date, time, guests))
    conn.commit()
    booking_id = cursor.lastrowid
    conn.close()
    return booking_id


def get_bookings_by_phone(phone: str) -> List[Dict[str, Any]]:
    """Получить брони по номеру телефона."""
    conn = get_db_connection()
    query = '''
        SELECT b.*, t.number as table_number, t.seats
        FROM bookings b
        JOIN tables t ON b.table_id = t.id
        WHERE b.customer_phone = ? AND b.status = 'active'
        ORDER BY b.date, b.time
    '''
    bookings = conn.execute(query, (phone,)).fetchall()
    conn.close()
    return [dict(row) for row in bookings]


def cancel_booking(booking_id: int, phone: str = None) -> bool:
    """Отменить бронь."""
    conn = get_db_connection()
    if phone:
        result = conn.execute('''
            UPDATE bookings SET status = 'cancelled'
            WHERE id = ? AND customer_phone = ? AND status = 'active'
        ''', (booking_id, phone))
    else:
        result = conn.execute('''
            UPDATE bookings SET status = 'cancelled'
            WHERE id = ? AND status = 'active'
        ''', (booking_id,))
    conn.commit()
    affected = result.rowcount
    conn.close()
    return affected > 0


def get_all_bookings() -> List[Dict[str, Any]]:
    """Получить все активные брони."""
    conn = get_db_connection()
    query = '''
        SELECT b.*, t.number as table_number
        FROM bookings b
        JOIN tables t ON b.table_id = t.id
        WHERE b.status = 'active'
        ORDER BY b.date, b.time
    '''
    bookings = conn.execute(query).fetchall()
    conn.close()
    return [dict(row) for row in bookings]