Урок 86 — Практикум 22: онлайн-магазин книг, часть 2 | Теория

После успешного входа пользователь попадает в функцию user_menu(connection, user_id). Это меню живёт отдельно от главного, потому что все его пункты связаны с уже авторизованным клиентом.

В меню доступны:

просмотр книг в наличии;
поиск книги по части названия;
покупка книги;
просмотр самых популярных поисковых запросов;
выход из аккаунта.

def user_menu(connection, user_id):
    while True:
        print(f"\n--- User Menu (ID {user_id}) ---")
        print("1. View available books")
        print("2. Search books by title")
        print("3. Purchase a book")
        print("4. View most frequent search queries")
        print("0. Logout")
        choice = input("Choose action: ")

        if choice == "1":
            view_available_books(connection)
        elif choice == "2":
            search_books_by_title(connection)
        elif choice == "3":
            purchase_book(connection, user_id)
        elif choice == "4":
            show_popular_queries()
        elif choice == "0":
            break
        else:
            print("Invalid option.")

📚 Просмотр и поиск книг

Для просмотра выбираем только книги с остатком больше нуля и нумеруем их для удобства выбора:

def view_available_books(connection):
    with connection.cursor() as cursor:
        cursor.execute("USE bookstore")
        cursor.execute("SELECT title, author, price, stock FROM books WHERE stock > 0")
        books = cursor.fetchall()

    if not books:
        print("No books available.")
        return

    print("Available books:")
    for i, (title, author, price, stock) in enumerate(books, start=1):
        print(f"{i}. {title} by {author} — ${price:.2f} ({stock} in stock)")

Поиск по названию работает регистронезависимо и тоже учитывает наличие на складе. Важно использовать параметризованный запрос, чтобы избежать SQL-инъекций:

def search_books_by_title(connection):
    keyword = input("Enter book title: ").strip()

    with connection.cursor() as cursor:
        cursor.execute("USE bookstore")
        cursor.execute("""
            SELECT title, author, price, stock
            FROM books
            WHERE stock > 0 AND LOWER(title) LIKE %s
        """, (f"%{keyword.lower()}%",))
        results = cursor.fetchall()

    if not results:
        print("No matching books found.")
        return

    print("Search results:")
    for i, (title, author, price, stock) in enumerate(results, start=1):
        print(f"{i}. {title} by {author} — ${price:.2f} ({stock} in stock)")

🛒 Покупка книги и таблица purchases

Покупка — это типичная бизнес-операция, которая затрагивает несколько таблиц. Алгоритм:

Показать доступные книги и получить от пользователя номер и количество.
Проверить, что номер книги корректен и количество не превышает остаток.
Получить баланс пользователя и проверить, что хватает денег.
Уменьшить остаток книги и баланс пользователя.
Записать покупку в таблицу purchases.
Сделать commit(), чтобы изменения сохранились.

def purchase_book(connection, user_id):
    with connection.cursor() as cursor:
        cursor.execute("USE bookstore")
        cursor.execute("SELECT id, title, author, price, stock FROM books WHERE stock > 0")
        books = cursor.fetchall()

    if not books:
        print("No books available.")
        return

    print("Available books:")
    for i, (book_id, title, author, price, stock) in enumerate(books, start=1):
        print(f"{i}. {title} by {author} — ${price:.2f} ({stock} in stock)")

    try:
        index = int(input("Enter book number: ")) - 1
        quantity = int(input("Enter quantity: "))
    except ValueError:
        print("Invalid input.")
        return

    if not (0 <= index < len(books)):
        print("Invalid book number.")
        return

    book_id, title, author, price, stock = books[index]

    if quantity > stock:
        print("Not enough books in stock.")
        return

    with connection.cursor() as cursor:
        cursor.execute("USE bookstore")
        cursor.execute("SELECT balance FROM users WHERE id = %s", (user_id,))
        balance = cursor.fetchone()[0]
        total_price = price * quantity

        if total_price > balance:
            print("Insufficient funds.")
            return

        cursor.execute(
            "UPDATE books SET stock = stock - %s WHERE id = %s",
            (quantity, book_id),
        )
        cursor.execute(
            "UPDATE users SET balance = balance - %s WHERE id = %s",
            (total_price, user_id),
        )
        cursor.execute("""
            INSERT INTO purchases (user_id, book_id, quantity, purchase_date)
            VALUES (%s, %s, %s, CURDATE())
        """, (user_id, book_id, quantity))
        connection.commit()
        print("Purchase successful.")

Таблица purchases связывает пользователей и книги внешними ключами:

purchases_query = """
    CREATE TABLE IF NOT EXISTS purchases (
        id INT AUTO_INCREMENT PRIMARY KEY,
        user_id INT,
        book_id INT,
        quantity INT,
        purchase_date DATE,
        FOREIGN KEY (user_id) REFERENCES users(id),
        FOREIGN KEY (book_id) REFERENCES books(id)
    )
"""

🍃 Логирование поисков в MongoDB

Реляционная база отлично подходит для строгих связей (пользователи, книги, покупки), а MongoDB — для гибких событий, таких как поисковые запросы. Каждый поиск сохраняем минимальным документом:

from pymongo import MongoClient

mongo_client = MongoClient(
    "mongodb://ich_editor:verystrongpassword"
    "@mongo.itcareerhub.de/?readPreference=primary"
    "&ssl=false&authMechanism=DEFAULT&authSource=ich_edit"
)
searches_collection = mongo_client["ich_edit"]["bookstore_logs_searches"]


def log_search_query(query: str) -> None:
    if query.strip():
        searches_collection.insert_one({"query": query.strip()})

📊 Анализ популярных запросов

Самый простой способ — загрузить все запросы в Python и посчитать через Counter:

from collections import Counter


def show_popular_queries(limit: int = 10) -> None:
    queries = [
        doc["query"].strip().lower()
        for doc in searches_collection.find()
        if doc.get("query")
    ]
    counter = Counter(queries)
    top = counter.most_common(limit)

    if not top:
        print("No search data found.")
        return

    print("Most frequent search queries:")
    for i, (query, count) in enumerate(top, start=1):
        print(f"{i}. {query} — {count} times")

Того же результата можно добиться прямо в MongoDB агрегацией $group — это удобно, когда записей много:

pipeline = [
    {"$match": {"query": {"$exists": True}}},
    {"$group": {"_id": {"$toLower": "$query"}, "count": {"$sum": 1}}},
    {"$sort": {"count": -1}},
    {"$limit": 10},
]
for doc in searches_collection.aggregate(pipeline):
    print(f"{doc['_id']} — {doc['count']} times")

📁 Разделение на модули

Когда в одном файле скапливается код для базы данных, авторизации, импорта книг, покупок и аналитики, его становится трудно читать и тестировать. Разделим программу на логические файлы:

bookstore_project/
├── main.py               # главное меню и точка входа
├── db_setup.py           # создание базы и таблиц
├── auth.py               # регистрация и вход
├── book_importer.py      # загрузка книг из CSV
├── user_actions.py       # просмотр, поиск, покупка
├── search_logs.py        # логирование и аналитика MongoDB
└── books.csv             # исходный каталог

В main.py остаются только импорты и запуск приложения:

from db_setup import create_database_and_tables
from auth import login_user, register_user
from book_importer import import_books_from_csv
from user_actions import user_menu


def main():
    connection = create_database_and_tables()
    import_books_from_csv(connection, "books.csv")

    while True:
        print("\n--- Bookstore ---")
        print("1. Register")
        print("2. Login")
        print("0. Exit")
        choice = input("Choose action: ")

        if choice == "1":
            register_user(connection)
        elif choice == "2":
            user_id = login_user(connection)
            if user_id:
                user_menu(connection, user_id)
        elif choice == "0":
            break


if __name__ == "__main__":
    main()

✨ Современные практики

Guard clauses. Сначала проверяйте ошибочные случаи и выходите из функции — код читается как « happy path» внизу.
Type hints. Аннотируйте параметры и возвращаемые значения (def f(connection, user_id: int) -> None).
Параметризованные запросы. Никогда не вставляйте пользовательский ввод в строку SQL через f"...{value}...".
Разделение ответственности. Один модуль — одна задача.
if __name__ == "__main__". Используйте, чтобы тестовый/запускающий код не выполнялся при импорте.
Переменные окружения для секретов. Логины и пароли не должны лежать в репозитории; храните их в .env и загружайте через python-dotenv.

⚠️ Проверить по документации: в лекции показано прямое подключение к учебному серверу MongoDB с паролем в коде. В рабочих проектах так делать нельзя — используйте переменные окружения.

📖 Теория: Практикум 22 — онлайн-магазин книг, часть 2