💻 Практические примеры

Пример 1. Создание базы данных

Подключаемся к серверу MySQL, создаём базу bookstore и проверяем, что она появилась в списке баз данных.

import pymysql

config = {
    "host": "ich-edit.edu.itcareerhub.de",
    "user": "ich1",
    "password": "ich1_password_ilovedbs",
}

db_name = "bookstore"

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        cursor.execute(f"CREATE DATABASE IF NOT EXISTS {db_name}")
        cursor.execute("SHOW DATABASES")
        databases = [row[0] for row in cursor]

        if db_name in databases:
            print(f"Database '{db_name}' created or already exists.")
        else:
            print("Something went wrong. Database not found.")

Что происходит:

1. pymysql.connect(**config) открывает соединение с сервером.
2. connection.cursor() создаёт курсор для выполнения SQL-команд.
3. CREATE DATABASE IF NOT EXISTS создаёт базу, если её ещё нет.
4. SHOW DATABASES возвращает список баз; мы проверяем, есть ли в нём bookstore.

Пример 2. Создание таблиц

После создания базы данных создаём таблицы books и users и выводим список таблиц.

import pymysql

config = {
    "host": "ich-edit.edu.itcareerhub.de",
    "user": "ich1",
    "password": "ich1_password_ilovedbs",
}

db_name = "bookstore"

books_query = """
    CREATE TABLE IF NOT EXISTS books (
        id INT AUTO_INCREMENT PRIMARY KEY,
        title VARCHAR(200),
        author VARCHAR(100),
        price DECIMAL(10, 2),
        stock INT CHECK (stock >= 0)
    )
"""

users_query = """
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        username VARCHAR(100),
        password VARCHAR(100),
        balance DECIMAL(10, 2) CHECK (balance >= 0)
    )
"""

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        cursor.execute(f"CREATE DATABASE IF NOT EXISTS {db_name}")
        cursor.execute(f"USE {db_name}")
        cursor.execute(books_query)
        cursor.execute(users_query)

        cursor.execute("SHOW TABLES")
        tables = [row[0] for row in cursor]

        print(f"Tables in '{db_name}':")
        for table in tables:
            print(f"- {table}")

Что происходит:

1. USE bookstore переключает курсор на нужную базу данных.
2. Выполняем многострочные SQL-запросы для создания таблиц.
3. SHOW TABLES возвращает список таблиц текущей базы.

Пример 3. Загрузка книг из CSV-файла

Читаем файл books.csv и добавляем книги в базу. Если книга с таким названием и автором уже есть — увеличиваем количество.

import pymysql

config = {
    "host": "ich-edit.edu.itcareerhub.de",
    "user": "ich1",
    "password": "ich1_password_ilovedbs",
}

db_name = "bookstore"


def load_books_from_file(filename: str, connection) -> None:
    added = 0
    with open(filename, encoding="utf-8") as file:
        with connection.cursor() as cursor:
            cursor.execute(f"USE {db_name}")
            for line in file:
                parts = line.strip().split(",")
                if len(parts) != 4:
                    continue
                title, author, price, stock = parts
                price = float(price)
                stock = int(stock)

                cursor.execute(
                    "SELECT id, stock FROM books WHERE title = %s AND author = %s",
                    (title, author),
                )
                result = cursor.fetchone()

                if result:
                    book_id, current_stock = result
                    cursor.execute(
                        "UPDATE books SET stock = %s WHERE id = %s",
                        (current_stock + stock, book_id),
                    )
                else:
                    cursor.execute(
                        "INSERT INTO books (title, author, price, stock) VALUES (%s, %s, %s, %s)",
                        (title, author, price, stock),
                    )
                added += 1
    connection.commit()
    print(f"{added} books loaded.")


with pymysql.connect(**config) as connection:
    filename = input("Enter file name: ")
    load_books_from_file(filename, connection)

Что происходит:

1. Файл открывается с кодировкой UTF-8 и читается построчно.
2. Каждая строка разбивается по запятым; некорректные строки пропускаются.
3. Цена и количество преобразуются к числовым типам.
4. Если книга найдена — обновляем остаток, иначе — вставляем новую запись.
5. connection.commit() фиксирует все изменения.

Пример файла books.csv:

1984,George Orwell,8.99,5
Brave New World,Aldous Huxley,9.50,3
To Kill a Mockingbird,Harper Lee,7.99,4

Пример 4. Регистрация пользователя

Запрашиваем логин, пароль и баланс. Если логин уже занят — отказываем в регистрации.

import pymysql

config = {
    "host": "ich-edit.edu.itcareerhub.de",
    "user": "ich1",
    "password": "ich1_password_ilovedbs",
}

db_name = "bookstore"


def register_user(connection) -> None:
    username = input("Enter username: ")
    password = input("Enter password: ")
    try:
        balance = float(input("Enter initial balance: "))
    except ValueError:
        print("Invalid balance.")
        return

    with connection.cursor() as cursor:
        cursor.execute(f"USE {db_name}")
        cursor.execute("SELECT id FROM users WHERE username = %s", (username,))
        if cursor.fetchone():
            print("Username already exists.")
            return

        cursor.execute(
            "INSERT INTO users (username, password, balance) VALUES (%s, %s, %s)",
            (username, password, balance),
        )
    connection.commit()
    print("Registration successful.")


with pymysql.connect(**config) as connection:
    register_user(connection)

Что происходит:

1. Проверяем корректность ввода баланса через try/except.
2. Ищем пользователя с таким же логином; если находим — выходим.
3. Вставляем нового пользователя параметризованным запросом.
4. Фиксируем транзакцию.

Пример 5. Вход в аккаунт и подменю

Проверяем логин и пароль. При успехе возвращаем user_id и открываем подменю клиента.

import pymysql

config = {
    "host": "ich-edit.edu.itcareerhub.de",
    "user": "ich1",
    "password": "ich1_password_ilovedbs",
}

db_name = "bookstore"


def try_login(connection) -> int | None:
    username = input("Enter username: ")
    password = input("Enter password: ")

    with connection.cursor() as cursor:
        cursor.execute(f"USE {db_name}")
        cursor.execute(
            "SELECT id FROM users WHERE username = %s AND password = %s",
            (username, password),
        )
        result = cursor.fetchone()

    if result:
        print("Login successful.")
        return result[0]
    print("Invalid username or password.")
    return None


def user_menu(connection, user_id: int) -> None:
    while True:
        print(f"\n--- User Menu (ID {user_id}) ---")
        print("1. View available books")
        print("0. Logout")
        choice = input("Choose action: ")

        if choice == "1":
            print("Books will be shown here.")
        elif choice == "0":
            break
        else:
            print("Invalid option.")


with pymysql.connect(**config) as connection:
    user_id = try_login(connection)
    if user_id:
        user_menu(connection, user_id)

Что происходит:

1. try_login возвращает либо идентификатор пользователя, либо None.
2. Подменю работает в отдельном цикле и завершается при выборе 0.
3. Выход из подменю возвращает управление в главное меню (в полной версии программы).

Пример 6. Полное главное меню

Объединяем все функции в одно главное меню с пунктами загрузки, регистрации, входа и выхода.

import pymysql

config = {
    "host": "ich-edit.edu.itcareerhub.de",
    "user": "ich1",
    "password": "ich1_password_ilovedbs",
}

db_name = "bookstore"


def load_books_from_file(filename: str, connection) -> None:
    added = 0
    with open(filename, encoding="utf-8") as file:
        with connection.cursor() as cursor:
            cursor.execute(f"USE {db_name}")
            for line in file:
                parts = line.strip().split(",")
                if len(parts) != 4:
                    continue
                title, author, price, stock = parts
                price = float(price)
                stock = int(stock)

                cursor.execute(
                    "SELECT id, stock FROM books WHERE title = %s AND author = %s",
                    (title, author),
                )
                result = cursor.fetchone()

                if result:
                    book_id, current_stock = result
                    cursor.execute(
                        "UPDATE books SET stock = %s WHERE id = %s",
                        (current_stock + stock, book_id),
                    )
                else:
                    cursor.execute(
                        "INSERT INTO books (title, author, price, stock) VALUES (%s, %s, %s, %s)",
                        (title, author, price, stock),
                    )
                added += 1
    connection.commit()
    print(f"{added} books loaded.")


def register_user(connection) -> None:
    username = input("Enter username: ")
    password = input("Enter password: ")
    try:
        balance = float(input("Enter initial balance: "))
    except ValueError:
        print("Invalid balance.")
        return

    with connection.cursor() as cursor:
        cursor.execute(f"USE {db_name}")
        cursor.execute("SELECT id FROM users WHERE username = %s", (username,))
        if cursor.fetchone():
            print("Username already exists.")
            return

        cursor.execute(
            "INSERT INTO users (username, password, balance) VALUES (%s, %s, %s)",
            (username, password, balance),
        )
    connection.commit()
    print("Registration successful.")


def try_login(connection) -> int | None:
    username = input("Enter username: ")
    password = input("Enter password: ")

    with connection.cursor() as cursor:
        cursor.execute(f"USE {db_name}")
        cursor.execute(
            "SELECT id FROM users WHERE username = %s AND password = %s",
            (username, password),
        )
        result = cursor.fetchone()

    if result:
        print("Login successful.")
        return result[0]
    print("Invalid username or password.")
    return None


def user_menu(connection, user_id: int) -> None:
    while True:
        print(f"\n--- User Menu (ID {user_id}) ---")
        print("1. View available books")
        print("0. Logout")
        choice = input("Choose action: ")

        if choice == "1":
            print("Books will be shown here.")
        elif choice == "0":
            break
        else:
            print("Invalid option.")


def main() -> None:
    with pymysql.connect(**config) as connection:
        while True:
            print("\n--- Bookstore Menu ---")
            print("1. Load books from file")
            print("2. Register new user")
            print("3. Login as user")
            print("0. Exit")
            choice = input("Choose action: ")

            if choice == "1":
                filename = input("Enter file name: ")
                load_books_from_file(filename, connection)
            elif choice == "2":
                register_user(connection)
            elif choice == "3":
                user_id = try_login(connection)
                if user_id:
                    user_menu(connection, user_id)
            elif choice == "0":
                break
            else:
                print("Invalid option.")


if __name__ == "__main__":
    main()

Что происходит:

1. Каждое действие вынесено в отдельную функцию.
2. Соединение открывается один раз и передаётся в функции.
3. Точка входа main() защищена конструкцией if __name__ == "__main__".

Как запустить в VS Code