💻 Примеры: итоговые сценарии

Пример 1. Подключение и простой SELECT

import pymysql

config = {
    "host": "ich-db.edu.itcareerhub.de",
    "user": "ich1",
    "password": "password",
    "database": "hr",
}

try:
    connection = pymysql.connect(**config)
    print("Connected:", connection.open)

    with connection.cursor() as cursor:
        cursor.execute("SELECT employee_id, first_name, last_name FROM employees LIMIT 5")
        for row in cursor:
            print(row)
finally:
    connection.close()

Здесь соединение создаётся из словаря конфигурации, запрос выполняется в контекстном менеджере with, а соединение закрывается в блоке finally.

Пример 2. Параметризованный запрос с %s

import pymysql

config = {
    "host": "ich-db.edu.itcareerhub.de",
    "user": "ich1",
    "password": "password",
    "database": "hr",
}

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        department_id = 60
        min_salary = 10000
        cursor.execute(
            "SELECT first_name, last_name, salary FROM employees "
            "WHERE department_id = %s AND salary > %s",
            (department_id, min_salary)
        )
        for row in cursor.fetchall():
            print(f"{row[0]} {row[1]}: {row[2]}")

Параметры передаются отдельно от SQL-запроса. Библиотека сама экранирует значения, защищая от SQL-инъекций.

Пример 3. Именованные параметры

import pymysql

config = {
    "host": "ich-db.edu.itcareerhub.de",
    "user": "ich1",
    "password": "password",
    "database": "hr",
}

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        params = {
            "department_id": 60,
            "min_salary": 10000,
        }
        cursor.execute(
            "SELECT first_name, last_name, salary FROM employees "
            "WHERE department_id = %(department_id)s AND salary > %(min_salary)s",
            params
        )
        for row in cursor:
            print(row)

Порядок ключей в словаре не важен — главное, чтобы имена совпадали с плейсхолдерами.

Пример 4. DictCursor

import pymysql
from pymysql.cursors import DictCursor

config = {
    "host": "ich-db.edu.itcareerhub.de",
    "user": "ich1",
    "password": "password",
    "database": "hr",
    "cursorclass": DictCursor,
}

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        cursor.execute(
            "SELECT first_name, last_name, job_id, salary FROM employees LIMIT 3"
        )
        for employee in cursor:
            print(f"{employee['first_name']} {employee['last_name']} "
                  f"works as {employee['job_id']} and earns {employee['salary']}")

С DictCursor каждая строка представляется словарём, и доступ к значениям идёт по именам колонок.

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

import pymysql

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

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        cursor.execute("CREATE DATABASE IF NOT EXISTS market")
        cursor.execute("USE market")
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS sales (
                id INT AUTO_INCREMENT PRIMARY KEY,
                item_name VARCHAR(100),
                quantity INT,
                price DECIMAL(10, 2),
                sale_date DATE
            )
        """)
        print("Database and table are ready.")

Для создания базы и таблиц подключаемся без указания database и используем учётную запись с правами на запись.

Пример 6. INSERT, UPDATE, DELETE с commit()

import pymysql

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

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        # Добавление
        cursor.execute(
            "INSERT INTO sales (item_name, quantity, price, sale_date) VALUES (%s, %s, %s, %s)",
            ("Keyboard", 2, 45.50, "2024-06-15")
        )

        # Обновление
        cursor.execute(
            "UPDATE sales SET quantity = %s WHERE item_name = %s",
            (3, "Keyboard")
        )

        # Удаление
        cursor.execute(
            "DELETE FROM sales WHERE item_name = %s",
            ("Keyboard",)
        )

    connection.commit()
    print("Changes saved.")

Изменяющие запросы требуют вызова commit(), чтобы данные сохранились в базе.

Пример 7. Массовая вставка через executemany()

import pymysql

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

with pymysql.connect(**config) as connection:
    with connection.cursor() as cursor:
        cursor.executemany(
            "INSERT INTO sales (item_name, quantity, price, sale_date) VALUES (%s, %s, %s, %s)",
            [
                ("Notebook", 3, 19.99, "2024-06-15"),
                ("Pen", 10, 1.99, "2024-06-16"),
                ("Bag", 1, 49.90, "2024-06-17"),
            ]
        )
    connection.commit()
    print("Bulk insert completed.")

executemany() выполняет один и тот же запрос для списка параметров. Это удобнее и эффективнее, чем вызывать execute() в цикле.

Пример 8. Транзакция покупки товара

import pymysql

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

def buy_product(customer_name, product_name):
    with pymysql.connect(**config) as connection:
        with connection.cursor() as cursor:
            try:
                # Получаем товар
                cursor.execute(
                    "SELECT id, price, stock FROM products WHERE name = %s",
                    (product_name,)
                )
                product = cursor.fetchone()
                if product is None:
                    raise ValueError("Product not found")
                product_id, price, stock = product

                if stock < 1:
                    raise ValueError("Out of stock")

                # Списываем товар со склада
                cursor.execute(
                    "UPDATE products SET stock = stock - 1 WHERE id = %s",
                    (product_id,)
                )

                # Получаем клиента
                cursor.execute(
                    "SELECT id, balance FROM customers WHERE name = %s",
                    (customer_name,)
                )
                customer = cursor.fetchone()
                if customer is None:
                    raise ValueError("Customer not found")
                customer_id, balance = customer

                if balance < price:
                    raise ValueError("Insufficient funds")

                # Списываем деньги и фиксируем покупку
                cursor.execute(
                    "UPDATE customers SET balance = balance - %s WHERE id = %s",
                    (price, customer_id)
                )
                cursor.execute(
                    "INSERT INTO purchases (customer_id, product_id, purchase_date) "
                    "VALUES (%s, %s, CURDATE())",
                    (customer_id, product_id)
                )

                connection.commit()
                print("Purchase successful.")
            except Exception as e:
                connection.rollback()
                print("Transaction failed:", e)


buy_product("Alice", "Headphones")
buy_product("Bob", "Mouse")

Если любой из шагов завершится ошибкой, rollback() отменит все изменения, и база останется в согласованном состоянии.