Как работать с MySQL из Python: пособие с примерами кода

Чтобы работать с MySQL из Python, поставьте драйвер mysql-connector-python командой pip и откройте соединение функцией connect(). Курсор выполняет SELECT, INSERT, UPDATE и DELETE, а commit() фиксирует изменения в базе. Вы получите рабочие примеры для каждой операции, хранимые процедуры, двоичные данные и пул соединений.

Обновлено: 2026-09-29 23:20:22 Сергей Бензенкоавтор материала

Что нужно для работы с MySQL из Python

Python сам не умеет говорить с сервером MySQL. Для этого нужен драйвер – библиотека, которая открывает соединение и отправляет SQL. Она же возвращает строки таблицы в виде типов Python. Значения драйвер переводит сам: datetime становится DATETIME и обратно.

Основных драйверов три. Таблица сравнивает их по данным страниц PyPI на момент обновления статьи.

Какой драйвер взять

mysql-connector-python – выбор для этой статьи. Его выпускает сам MySQL, а установка сводится к одной команде pip. На PyPI сейчас опубликована версия 26.7.0.

У драйвера есть C-расширение: параметр use_pure по умолчанию равен False, и коннектор берёт расширение, если оно доступно. Чтобы принудительно работать на чистом Python, передайте use_pure=True.

Берите PyMySQL, когда нужен минимум зависимостей. Это чистый Python под лицензией MIT. В его connect() параметры называются password и database, а прежние passwd и db устарели.

Не начинайте с mysqlclient, если вам не нужен именно он. Для установки на Linux нужны заголовки клиентской библиотеки и компилятор, а для Windows есть готовые бинарные колёса. Исключение – Django: его документация называет mysqlclient рекомендуемым драйвером.

Установка драйвера и тестовая база

Установка драйвера

Ставьте драйвер в виртуальное окружение проекта. Так он не смешается с системным Python и с другими проектами. Драйверу нужен Python 3.10 или новее: свою версию покажет команда python --version. В Linux и macOS команда часто называется python3, тогда запустите python3 --version.

1. Откройте терминал в папке проекта: PowerShell или «Командную строку» в Windows, любой терминал в Linux и macOS.
2. Создайте окружение командой python -m venv .venv. Если система отвечает «command not found», используйте python3 -m venv .venv. В папке появится каталог .venv.
3. Активируйте окружение: в Windows командой .venv\Scripts\activate, в Linux и macOS командой source .venv/bin/activate. В начале строки терминала появится «(.venv)».
4. Установите драйвер командой pip install mysql-connector-python.
5. Установите читалку файла .env командой pip install python-dotenv. Она понадобится в разделе про подключение.
6. Проверьте установку командой pip show mysql-connector-python. В ответе появятся строки «Name» и «Version».

Если PowerShell отказывается активировать окружение, запустите ту же команду в «Командной строке».

Тестовая база для примеров

Все примеры работают с базой python_mysql и тремя таблицами: книги, авторы и связь между ними. Откройте MySQL Workbench или консольный клиент (mysql -u root -p) и выполните скрипт целиком.

CREATE DATABASE python_mysql CHARACTER SET utf8mb4;
USE python_mysql;

CREATE TABLE books (
  id INT AUTO_INCREMENT PRIMARY KEY,
  title VARCHAR(255) NOT NULL,
  isbn VARCHAR(13)
) ENGINE=InnoDB;

CREATE TABLE authors (
  id INT AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(50) NOT NULL,
  last_name VARCHAR(50) NOT NULL,
  photo MEDIUMBLOB
) ENGINE=InnoDB;

CREATE TABLE book_author (
  book_id INT NOT NULL,
  author_id INT NOT NULL,
  PRIMARY KEY (book_id, author_id),
  FOREIGN KEY (book_id) REFERENCES books (id) ON DELETE CASCADE,
  FOREIGN KEY (author_id) REFERENCES authors (id) ON DELETE CASCADE
) ENGINE=InnoDB;

INSERT INTO books (id, title, isbn) VALUES
  (37, 'The Giant on the Hill', '1236400967773'),
  (102, 'Book To Delete', '9780000000102');
INSERT INTO authors (id, first_name, last_name) VALUES (144, 'Garth', 'Stein');
INSERT INTO book_author (book_id, author_id) VALUES (37, 144), (102, 144);

Идентификаторы 37, 102 и 144 заданы явно: на них ссылаются примеры ниже. Движок InnoDB поддерживает транзакции и внешние ключи, поэтому его указали в каждой таблице. Столбец isbn рассчитан на 13 цифр без дефисов.

Для проверки хватит пользователя root. В рабочем проекте заведите отдельного пользователя с правами только на эту базу: при утечке пароля пострадает одна база, а не весь сервер.

Подключение к базе данных

Параметры connect()

Функция mysql.connector.connect() открывает соединение и возвращает объект соединения. Для начала хватит четырёх параметров: host, user, password и database. Параметр use_pure, о котором шла речь выше, принимает именно connect(), поэтому все примеры статьи открывают соединение через неё.

Метод is_connected() проверяет, что связь жива. Если сервер недоступен, база не существует или пароль неверный, коннектор бросает исключение mysql.connector.Error. Ловите именно его.

Доступы не хранят в коде

Строка password='secret' в скрипте рано или поздно попадает в репозиторий. Берите доступы из переменных окружения. Функция os.getenv() возвращает None, если переменной нет, и это легко проверить.

Для разработки удобен файл .env рядом со скриптом:

MYSQL_HOST=localhost
MYSQL_USER=root
MYSQL_PASSWORD=ваш_пароль
MYSQL_DATABASE=python_mysql

Добавьте .env в .gitignore, чтобы пароль не уехал в репозиторий. Библиотека python-dotenv читает этот файл в os.environ: она ищет .env в папке скрипта и выше по дереву.

Без файла задайте переменные в терминале: set MYSQL_PASSWORD=... в «Командной строке», $env:MYSQL_PASSWORD="..." в PowerShell, export MYSQL_PASSWORD=... в Linux и macOS.

Сохраните общий модуль как db.py. Остальные примеры импортируют из него функцию connect(), поэтому кладите все файлы в одну папку.

import os

import mysql.connector
from dotenv import load_dotenv

load_dotenv()


def get_config():
    password = os.getenv("MYSQL_PASSWORD")
    if password is None:
        raise RuntimeError("Задайте переменную окружения MYSQL_PASSWORD")
    return {
        "host": os.getenv("MYSQL_HOST", "localhost"),
        "user": os.getenv("MYSQL_USER", "root"),
        "password": password,
        "database": os.getenv("MYSQL_DATABASE", "python_mysql"),
    }


def connect():
    return mysql.connector.connect(**get_config())

Закрытие соединения без NameError

Во многих учебниках соединение открывают внутри try, а закрывают в finally. Если connect() упал, переменная conn не создана. Блок finally тогда падает с NameError и прячет настоящую причину – например, неверный пароль.

Функция closing() из стандартной библиотеки решает эту задачу. Блок with closing(connect()) as conn закрывает соединение, только если оно открылось. Ошибка подключения уходит наружу как есть. Тот же приём годится для курсора.

Сохраните листинг как check.py в той же папке, что и db.py.

from contextlib import closing

from mysql.connector import Error

from db import connect


def check():
    try:
        with closing(connect()) as conn:
            print("Подключено:", conn.is_connected())
    except Error as e:
        print("Ошибка подключения:", e)


if __name__ == "__main__":
    check()

Запустите python check.py. На экране появится строка «Подключено: True». Если вместо неё стоит сообщение об ошибке, разбор частых причин – в конце статьи.

Курсор и выборка данных

Курсор – объект, через который вы отправляете SQL и забираете строки. Любая работа с таблицей идёт в одном порядке:

1. Откройте соединение функцией connect().
2. Создайте курсор методом conn.cursor().
3. Выполните запрос: cursor.execute(query, params).
4. Заберите строки методом fetchone(), fetchmany() или fetchall().
5. Закройте курсор и соединение блоками with closing(...).

В следующих примерах меняются только запрос и то, что вы делаете с результатом.

fetchone: строка за строкой

Метод fetchone() возвращает следующую строку результата или None, когда строки кончились. Строка приходит кортежем, например (37, 'The Giant on the Hill', '1236400967773').

from contextlib import closing

from db import connect


def query_with_fetchone():
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute("SELECT id, title, isbn FROM books")
            row = cursor.fetchone()
            while row is not None:
                print(row)
                row = cursor.fetchone()


if __name__ == "__main__":
    query_with_fetchone()

fetchall: весь результат сразу

Метод fetchall() возвращает список всех оставшихся строк. Берите его, когда результат небольшой: справочник, настройки, список из нескольких десятков записей. Весь результат при этом лежит в памяти программы.

from contextlib import closing

from db import connect


def query_with_fetchall():
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute("SELECT id, title, isbn FROM books")
            rows = cursor.fetchall()

    print("Всего строк:", len(rows))
    for row in rows:
        print(row)


if __name__ == "__main__":
    query_with_fetchall()

fetchmany: порциями

Большая таблица в fetchall() займёт много памяти. Метод fetchmany(n) отдаёт до n строк за вызов и пустой список в конце. Генератор iter_row() прячет эти вызовы: цикл for видит обычную последовательность строк.

from contextlib import closing

from db import connect


def iter_row(cursor, size=10):
    while True:
        rows = cursor.fetchmany(size)
        if not rows:
            break
        for row in rows:
            yield row


def query_with_fetchmany():
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute("SELECT id, title, isbn FROM books")
            for row in iter_row(cursor, 10):
                print(row)


if __name__ == "__main__":
    query_with_fetchmany()

Курсор с параметром buffered=True забирает с сервера все строки сразу. Тогда выигрыш fetchmany() по памяти пропадает, поэтому для больших выборок этот параметр не включайте.

Строки словарями: dictionary=True

Кортеж неудобен: столбец приходится искать по номеру. Параметр cursor(dictionary=True) отдаёт каждую строку словарём, где ключ – имя столбца.

from contextlib import closing

from db import connect


def get_book(book_id):
    with closing(connect()) as conn:
        with closing(conn.cursor(dictionary=True)) as cursor:
            cursor.execute("SELECT id, title FROM books WHERE id = %s", (book_id,))
            return cursor.fetchone()


if __name__ == "__main__":
    book = get_book(37)
    print(book["title"])

Какой метод выбрать

Вставка данных

Одна строка и её идентификатор

Оператор INSERT принимает значения через заполнители %s. Это не форматирование строк Python: драйвер сам подставляет значения в запрос безопасно. Заполнитель всегда %s, даже для чисел и дат, а кавычек вокруг него нет. Значения передают кортежем вторым аргументом execute().

Свойство cursor.lastrowid хранит значение AUTO_INCREMENT последней вставки. Вызов conn.commit() обязателен: без него строка в таблице не появится.

from contextlib import closing

from db import connect


def insert_book(title, isbn):
    query = "INSERT INTO books (title, isbn) VALUES (%s, %s)"
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(query, (title, isbn))
            conn.commit()
            return cursor.lastrowid


if __name__ == "__main__":
    new_id = insert_book("A Sudden Light", "9781439187036")
    print("Новая книга получила id", new_id)

Несколько строк: executemany

В самом SQL несколько строк вставляются одним оператором со списками значений:

INSERT INTO books (title, isbn) VALUES
  ('Harry Potter And The Order Of The Phoenix', '9780439358071'),
  ('Gone with the Wind', '9780446675536'),
  ('Pride and Prejudice (Modern Library Classics)', '9780679783268');

В Python для этого есть executemany(). Вы передаёте один запрос и список кортежей. Для INSERT коннектор сам объединяет значения в оператор с несколькими строками.

from contextlib import closing

from db import connect


def insert_books(books):
    query = "INSERT INTO books (title, isbn) VALUES (%s, %s)"
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.executemany(query, books)
        conn.commit()


if __name__ == "__main__":
    insert_books([
        ("Harry Potter And The Order Of The Phoenix", "9780439358071"),
        ("Gone with the Wind", "9780446675536"),
        ("Pride and Prejudice (Modern Library Classics)", "9780679783268"),
    ])

Фиксация одна на всю пачку: три книги попадают в базу вместе.

Обновление и удаление данных

Без commit() UPDATE и DELETE ничего не меняют в базе. Скрипт завершится без ошибки, а таблица останется прежней. Вызывайте commit() после каждого изменения, которое должно сохраниться.

Обновление строки

Функция update_book() меняет название книги по её id. Условие WHERE id = %s не даёт затронуть остальные строки. Сохраните листинг как update_book.py.

from contextlib import closing

from db import connect


def update_book(book_id, title):
    query = """
        UPDATE books
        SET title = %s
        WHERE id = %s
    """
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(query, (title, book_id))
        conn.commit()


if __name__ == "__main__":
    update_book(37, "The Giant on the Hill *** TEST ***")

Проверьте результат в три шага. Сначала выберите книгу в клиенте MySQL: SELECT * FROM books WHERE id = 37;. Скриншот показывает, как выглядит такая выборка в клиенте; значения на нём могут отличаться от вашей базы.

Затем запустите скрипт командой python update_book.py. После этого повторите тот же SELECT. В названии появится «*** TEST ***».

Удаление строки

Оператор DELETE без WHERE удаляет все строки таблицы. Всегда задавайте условие и сначала проверяйте его обычным SELECT.

Удалим книгу с id 102. Строка из book_author уйдёт вместе с ней: внешний ключ создан с ON DELETE CASCADE. Сохраните листинг как delete_book.py.

from contextlib import closing

from db import connect


def delete_book(book_id):
    query = "DELETE FROM books WHERE id = %s"
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(query, (book_id,))
        conn.commit()


if __name__ == "__main__":
    delete_book(102)

Значение book_id передаётся кортежем из одного элемента: (book_id,). Запятая обязательна, иначе Python прочтёт скобки как обычные. Перед запуском выполните SELECT * FROM books WHERE id = 102; – строка есть. Скриншот ниже показывает вид такой проверки; данные на нём могут отличаться от вашей базы.

Запустите python delete_book.py и повторите SELECT. Запрос вернёт пустой результат: строка удалена.

Транзакции и обработка ошибок

Транзакция объединяет несколько запросов в одно целое: сохраняются все или ни один. Пример – книга и её автор. Если запись в book_author упадёт, книга без автора остаться не должна.

Схема простая. Запросы выполняются один за другим, а commit() стоит в самом конце. Любая ошибка ведёт к rollback(), и все изменения отменяются.

from contextlib import closing

from mysql.connector import Error

from db import connect


def add_book_with_author(title, isbn, author_id):
    conn = None
    try:
        conn = connect()
        with closing(conn.cursor()) as cursor:
            cursor.execute(
                "INSERT INTO books (title, isbn) VALUES (%s, %s)", (title, isbn)
            )
            book_id = cursor.lastrowid
            cursor.execute(
                "INSERT INTO book_author (book_id, author_id) VALUES (%s, %s)",
                (book_id, author_id),
            )
        conn.commit()
        print("Книга добавлена, id", book_id)
    except Error as e:
        if conn is not None and conn.is_connected():
            conn.rollback()
        print("Транзакция отменена:", e)
    finally:
        if conn is not None and conn.is_connected():
            conn.close()


if __name__ == "__main__":
    add_book_with_author("Book Without Author", "9780000000003", 99999)

Автора с id 99999 в базе нет. Внешний ключ отклонит вторую вставку, сработает rollback(), и книга не попадёт в таблицу. Замените 99999 на 144, и обе строки сохранятся.

Зачем проверки перед rollback() и close()

Соединение может не открыться или оборваться посреди работы. Тогда conn равно None или уже недоступно. Вызов conn.rollback() у None даёт AttributeError поверх первой ошибки, а вызов на оборванном соединении способен дать новую. Проверка conn is not None and conn.is_connected() оставляет на экране одну, настоящую причину.

Когда транзакция не нужна, хватает блоков with closing(...) из прошлых разделов. Их проще читать, но откат вы в них не делаете.

Хранимые процедуры

Создание двух процедур

Хранимая процедура – набор SQL-команд, который лежит на сервере и вызывается по имени. Создадим две: find_all возвращает все книги с авторами, find_by_isbn ищет название по ISBN.

Команду DELIMITER понимает клиент MySQL и Workbench, а коннектору её передавать не нужно. Выполните скрипт в клиенте, не в Python.

USE python_mysql;

DELIMITER $$

CREATE PROCEDURE find_all()
BEGIN
  SELECT b.title, b.isbn, CONCAT(a.first_name, ' ', a.last_name) AS author
  FROM books AS b
  INNER JOIN book_author AS ba ON ba.book_id = b.id
  INNER JOIN authors AS a ON ba.author_id = a.id;
END$$

CREATE PROCEDURE find_by_isbn(IN p_isbn VARCHAR(13), OUT p_title VARCHAR(255))
BEGIN
  SELECT title INTO p_title FROM books WHERE isbn = p_isbn;
END$$

DELIMITER ;

Проверьте вторую процедуру в клиенте: CALL find_by_isbn('1236400967773', @title);, затем SELECT @title;. Вернётся название книги с этим ISBN. Если вы уже запускали update_book.py, в нём будет «*** TEST ***».

Вызов через callproc()

Метод cursor.callproc(name, args) принимает имя процедуры и список параметров. Набор строк, который вернула процедура, читают методом cursor.stored_results(): он отдаёт итератор по результатам, и у каждого есть fetchall().

Метод stored_results() объявлен устаревшим с версии 9.3.0. В примечаниях к этому выпуску разработчики называют его «пока просто устаревшим». Листинг ниже использует его, поэтому перед обновлением коннектора читайте примечания к выпуску.

Если метод уберут, вместо процедуры отправляйте обычный SELECT через execute(), а одиночные значения получайте через параметры OUT, как в следующем примере.

from contextlib import closing

from db import connect


def call_find_all():
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.callproc("find_all")
            for result in cursor.stored_results():
                for row in result.fetchall():
                    print(row)


if __name__ == "__main__":
    call_find_all()

Параметр OUT

Процедура find_by_isbn отдаёт название через параметр OUT. В списке аргументов на его место ставят заглушку, например 0. Метод callproc() возвращает изменённую копию этого списка: входные параметры остаются прежними, а на месте OUT стоит значение из процедуры.

from contextlib import closing

from db import connect


def call_find_by_isbn(isbn):
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            result_args = cursor.callproc("find_by_isbn", [isbn, 0])
            return result_args[1]


if __name__ == "__main__":
    print(call_find_by_isbn("1236400967773"))

Индекс 1 указывает на второй элемент списка – параметр p_title. Для других процедур считайте позицию OUT-параметра в её объявлении.

Двоичные данные (BLOB)

BLOB – тип столбца для двоичных данных: картинок, архивов, документов. У него три размера:

• BLOB – до 65 535 байт;
• MEDIUMBLOB – до 16 777 215 байт;
• LONGBLOB – до 4 294 967 295 байт, около 4 ГБ.

Столбец authors.photo из нашего скрипта имеет тип MEDIUMBLOB, то есть держит файл до 16 МБ. Ограничивает и сервер: параметр max_allowed_packet по умолчанию равен 64 МБ. Пакет крупнее закончится ошибкой «Got a packet bigger than 'max_allowed_packet' bytes» (код 1153).

Запись файла в столбец

Создайте папку pictures рядом со скриптом. Подойдёт любой JPG-файл: положите его туда и назовите garth_stein.jpg. Именно это имя использует пример.

Функция read_file() читает файл в байты. Функция update_blob() отправляет их в запрос как обычный параметр %s. Пути строятся от файла скрипта через pathlib, поэтому обратные слэши Windows не съедаются.

from contextlib import closing
from pathlib import Path

from db import connect

BASE = Path(__file__).resolve().parent


def read_file(path):
    return Path(path).read_bytes()


def update_blob(author_id, path):
    data = read_file(path)
    query = "UPDATE authors SET photo = %s WHERE id = %s"
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(query, (data, author_id))
        conn.commit()


if __name__ == "__main__":
    update_blob(144, BASE / "pictures" / "garth_stein.jpg")

Проверьте запись в клиенте MySQL: SELECT * FROM authors WHERE id = 144;. В столбце photo появятся двоичные данные вместо NULL.

Чтение обратно в файл

Функция read_blob() выбирает столбец photo и записывает байты в файл. Она сначала проверяет, что строка найдена и столбец не пуст: иначе fetchone()[0] упадёт на None.

from contextlib import closing
from pathlib import Path

from db import connect

BASE = Path(__file__).resolve().parent


def write_file(data, path):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)


def read_blob(author_id, path):
    query = "SELECT photo FROM authors WHERE id = %s"
    with closing(connect()) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(query, (author_id,))
            row = cursor.fetchone()

    if row is None or row[0] is None:
        raise LookupError(f"У автора {author_id} нет фото")
    write_file(row[0], path)


if __name__ == "__main__":
    read_blob(144, BASE / "output" / "garth_stein.jpg")

Откройте папку output рядом со скриптом. Файл garth_stein.jpg открывается как обычная картинка – значит, данные прошли туда и обратно без потерь.

Не храните в BLOB большие файлы. Каждый SELECT * тянет двоичные данные по сети, и таблица быстро раздувается. Файлы крупнее нескольких мегабайт держите на диске или в объектном хранилище, а в таблице оставьте путь.

Пул соединений

Открытие соединения – это сетевое рукопожатие и проверка пароля. В веб-приложении такое соединение на каждый запрос стоит времени и лишней нагрузки на сервер. Пул решает эту задачу: соединения открываются заранее, запросы берут готовое и возвращают его обратно.

Класс mysql.connector.pooling.MySQLConnectionPool принимает pool_name, pool_size и pool_reset_session. Размер по умолчанию – 5 соединений, максимум – 32. Остальные параметры, включая доступы, уходят в connect() через **kwargs.

from contextlib import closing

from mysql.connector import pooling

from db import get_config

pool = pooling.MySQLConnectionPool(
    pool_name="books_pool",
    pool_size=5,
    **get_config(),
)


def count_books():
    conn = pool.get_connection()
    try:
        with closing(conn.cursor()) as cursor:
            cursor.execute("SELECT COUNT(*) FROM books")
            return cursor.fetchone()[0]
    finally:
        conn.close()


if __name__ == "__main__":
    print("Книг в таблице:", count_books())

Создавайте пул один раз при старте приложения, а не в каждой функции. Метод get_connection() отдаёт свободное соединение из пула. Вызов close() на нём не разрывает связь с сервером: соединение возвращается в пул и ждёт следующего запроса. Забыли вернуть – пул из 5 соединений опустеет после пяти таких запросов.

Пул создают и без класса: вызов mysql.connector.connect(pool_name="books_pool", pool_size=5, ...) тоже открывает пул. Размер подбирайте под число одновременных запросов вашего приложения, а не «с запасом».

Частые ошибки

Ошибки подключения: 1045, 2003 и 1049

Ошибки подключения приходят как mysql.connector.Error с номером и текстом. Три встречаются чаще других.

Плагин входа в MySQL 8.4

Начиная с MySQL 8.0 по умолчанию работает плагин caching_sha2_password. В MySQL 8.4 плагин mysql_native_password отключён по умолчанию: сервер включает его только с опцией --mysql-native-password=ON. Если вы переносите старую учётную запись на 8.4, сначала проверьте, каким плагином она создана. Ошибку у такой записи ищите в тексте исключения, а не по номеру из таблицы выше.

Забытый commit

Скрипт завершился без ошибок, а данные прежние. Почти всегда причина одна: после INSERT, UPDATE или DELETE не вызван conn.commit(). Найдите в коде каждый изменяющий запрос и убедитесь, что после него стоит фиксация.

SQL-инъекция при склейке строк

Если вклеить значение в запрос f-строкой, злоумышленник дописывает в него собственный SQL. Значение x' OR '1'='1 превращает условие в «всегда истинно», и запрос вернёт всю таблицу. Скрипт ниже сравнивает два способа: первый запрос вернёт все книги, второй – ни одной.

from contextlib import closing

from db import connect

title = "x' OR '1'='1"

with closing(connect()) as conn:
    with closing(conn.cursor()) as cursor:
        # Опасно: значение вклеено в текст запроса
        cursor.execute(f"SELECT id, title FROM books WHERE title = '{title}'")
        print("Склейка:", len(cursor.fetchall()), "строк")

        # Безопасно: значение передано отдельно от запроса
        cursor.execute("SELECT id, title FROM books WHERE title = %s", (title,))
        print("Параметр:", len(cursor.fetchall()), "строк")

Не собирайте SQL из данных пользователя ни f-строкой, ни конкатенацией, ни оператором %. Передавайте значения вторым аргументом execute(): заполнитель %s работает для любого типа.

Совет от нашей редакции: начните с двух привычек – параметры через %s и явный commit(). Они закрывают самые дорогие ошибки, а остальные видны по тексту исключения.