💻 Примеры: покупка, логирование, аналитика, модули

💻 Пример 1. Покупка книги с проверкой баланса

Функция сначала проверяет «неправильные» сценарии (guard clauses), а затем выполняет обновления и фиксирует покупку.

import pymysql


def purchase_book(connection, user_id: int) -> None:
    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

    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.")

💻 Пример 2. Сохранение поиска в 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:
    clean = query.strip()
    if clean:
        searches_collection.insert_one({"query": clean})


def search_books_by_title(connection) -> None:
    keyword = input("Enter book title: ").strip()
    log_search_query(keyword)

    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)")

💻 Пример 3. Топ популярных запросов через 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")

💻 Пример 4. Тот же топ через MongoDB aggregation

def show_popular_queries_aggregate(limit: int = 10) -> None:
    pipeline = [
        {"$match": {"query": {"$exists": True}}},
        {"$group": {"_id": {"$toLower": "$query"}, "count": {"$sum": 1}}},
        {"$sort": {"count": -1, "_id": 1}},
        {"$limit": limit},
    ]

    results = list(searches_collection.aggregate(pipeline))
    if not results:
        print("No search data found.")
        return

    print("Most frequent search queries:")
    for i, doc in enumerate(results, start=1):
        print(f"{i}. {doc['_id']} — {doc['count']} times")

💻 Пример 5. Модульная структура проекта

Файл user_actions.py отвечает только за действия клиента. Он импортирует логирование из search_logs.py:

# user_actions.py
import pymysql
from search_logs import log_search_query


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)")


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

    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)")

Файл main.py только запускает приложение:

# 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()