Работа с БД

Всё, что программа держит в переменных и объектах, исчезает вместе с процессом: Registrar из CLI в статье про декораторы каждый запуск начинал с чистого листа
База данных переживает завершение процесса и даёт нескольким программам общие данные
В этой статье - SQLite (база в одном файле, модуль sqlite3 уже входит в Python) и коротко PostgreSQL через psycopg

Теория

Реляционная модель в двух словах

Реляционная база хранит данные в таблицах: колонки задают имя и тип поля, строки - отдельные записи
Строку однозначно определяет первичный ключ (PRIMARY KEY) - колонка или набор колонок, значения которых не повторяются
Связь между таблицами задаёт внешний ключ (FOREIGN KEY) - колонка, значения которой обязаны существовать в первичном ключе другой таблицы

Для сквозной модели курсов и студентов нужны три таблицы:

• students - студенты: student_id (первичный ключ) и name
• lessons - занятия: lesson_id (первичный ключ), kind (лекция или семинар) и capacity (число мест)
• enrollments - записи на занятия: пара student_id и lesson_id как составной первичный ключ, каждая половина пары - внешний ключ на свою таблицу

Таблица enrollments нужна из-за связи «многие ко многим» (студент ходит на много занятий, на занятие ходит много студентов): такую связь нельзя положить ни в одну из двух таблиц, её хранят отдельной таблицей-связкой

SQLite и модуль sqlite3

SQLite - база данных в одном файле, которая работает внутри самого процесса: сервер не нужен, ничего не нужно устанавливать, а модуль sqlite3 входит в стандартную библиотеку Python
Поэтому SQLite удобна для скриптов, тестов и небольших приложений, а для сервисов с большим числом одновременно пишущих клиентов берут серверную базу вроде PostgreSQL

import sqlite3

conn = sqlite3.connect(":memory:")  # база в памяти, пропадёт вместе с процессом
conn.execute(
    """
    CREATE TABLE students (
        student_id TEXT NOT NULL PRIMARY KEY,
        name       TEXT NOT NULL
    )
    """
)
conn.execute("INSERT INTO students (student_id, name) VALUES ('S1', 'Алиса')")
conn.commit()

cursor = conn.execute("SELECT student_id, name FROM students")
print(cursor.fetchall())  # [('S1', 'Алиса')]
conn.close()

sqlite3.connect(path) открывает файл базы или создаёт его, а ":memory:" вместо пути даёт базу в памяти - удобно для экспериментов и тестов
conn.execute(sql) выполняет запрос и возвращает курсор: fetchone() отдаёт одну строку (или None, если строк не осталось), а fetchall() - список всех оставшихся строк
По умолчанию строка - это кортеж
Часть возможностей зависит от версии самой библиотеки SQLite, её показывает sqlite3.sqlite_version
Что значит каждая строчка CREATE TABLE и остальных запросов, разобрано в разделе «Основные запросы SQL» ниже

Типы данных

SQLite хранит значения пяти видов, и при чтении sqlite3 превращает их в обычные типы Python:

• INTEGER - целое число, в Python это int
• REAL - число с плавающей точкой, в Python это float
• TEXT - строка, в Python это str
• BLOB - произвольные байты, в Python это bytes
• NULL - «значения нет», в Python это None

Тип колонки в CREATE TABLE - скорее подсказка, чем жёсткое правило: в колонку INTEGER можно вставить строку 'abc', и SQLite молча сохранит её как текст
Если нужна строгая проверка, таблицу создают с ключевым словом STRICT (SQLite 3.37 и новее): тогда та же вставка упадёт с sqlite3.IntegrityError

CREATE TABLE ages (age INTEGER) STRICT;

Основные запросы SQL

SQL - язык, на котором программа общается с реляционной базой: запрос описывает, какие данные нужны, а как их достать, решает сама база
Запросы делятся на два вида: описание структуры (CREATE TABLE) и работа с данными (INSERT, SELECT, UPDATE, DELETE)

Несколько общих правил:

• ключевые слова не зависят от регистра, но принято писать их заглавными
• строки пишутся в одинарных кавычках: 'Алиса'
• запрос заканчивается точкой с запятой, а при вызове execute() из Python её обычно опускают
• комментарий начинается с -- и идёт до конца строки

В примерах ниже используются таблицы students, lessons и enrollments, а результаты запросов записаны комментариями под ними

CREATE TABLE - описать таблицу

CREATE TABLE IF NOT EXISTS lessons (
    lesson_id TEXT NOT NULL PRIMARY KEY,
    kind      TEXT NOT NULL CHECK (kind IN ('lecture', 'seminar')),
    capacity  INTEGER NOT NULL CHECK (capacity > 0)
);

После имени таблицы в скобках идёт список колонок через запятую: у каждой есть имя, тип и ограничения (constraints) - правила, которые база проверяет при каждой записи
IF NOT EXISTS значит «создать, только если такой таблицы ещё нет»: без него повторный запуск скрипта упадёт с sqlite3.OperationalError (таблица уже существует)

Ограничения, которые встречаются чаще всего:

• PRIMARY KEY - первичный ключ: значения не повторяются (NOT NULL рядом нужен из-за особенности SQLite: у текстовых ключей без него допустим NULL, а PostgreSQL такого не разрешает)
• NOT NULL - колонка обязана иметь значение, NULL вставить нельзя
• UNIQUE - значения в колонке не повторяются, как у первичного ключа, но таких колонок в таблице может быть несколько (например, email TEXT UNIQUE)
• DEFAULT - значение по умолчанию, которое подставится, если при INSERT колонку не указали (например, status TEXT NOT NULL DEFAULT 'active')
• CHECK (условие) - значение должно проходить условие: capacity > 0 не даст создать занятие без мест
• REFERENCES таблица (колонка) - внешний ключ: значение обязано существовать в указанной колонке другой таблицы

Внешние ключи и составной первичный ключ - в таблице записей:

CREATE TABLE IF NOT EXISTS enrollments (
    student_id TEXT NOT NULL REFERENCES students (student_id),
    lesson_id  TEXT NOT NULL REFERENCES lessons (lesson_id),
    PRIMARY KEY (student_id, lesson_id)
);

Строка PRIMARY KEY (student_id, lesson_id) в конце - составной первичный ключ: не повторяться должна пара значений целиком
Поэтому студент может ходить на много занятий, а на одно и то же занятие записаться дважды не может

Числовой первичный ключ обычно не придумывают самим, а поручают базе
В SQLite колонка, объявленная ровно как INTEGER PRIMARY KEY, нумеруется автоматически: если при INSERT её не указать, база подставит число на единицу больше самого большого из существующих (в пустой таблице - 1):

CREATE TABLE IF NOT EXISTS notes (
    note_id INTEGER PRIMARY KEY,
    text    TEXT NOT NULL
);

INSERT INTO notes (text) VALUES ('первая'), ('вторая');

SELECT note_id, text FROM notes;
-- 1 | первая
-- 2 | вторая

Тип должен называться именно INTEGER: у колонки INT PRIMARY KEY автонумерации нет, и номер остаётся NULL
Если удалить последнюю строку, её номер получит следующая вставка, а чтобы номера не использовались повторно, после PRIMARY KEY пишут AUTOINCREMENT
В PostgreSQL то же самое записывают иначе: note_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY

INSERT - добавить строки

INSERT INTO students (student_id, name) VALUES ('S1', 'Алиса');

В скобках после имени таблицы перечислены колонки, а в VALUES - значения в том же порядке
Колонки лучше называть явно: запрос не сломается, если в таблицу позже добавят ещё одну

За один запрос можно вставить несколько строк, перечислив наборы значений через запятую:

INSERT INTO students (student_id, name) VALUES ('S2', 'Боря'), ('S3', 'Вера');

INSERT INTO lessons (lesson_id, kind, capacity)
VALUES ('L1', 'lecture', 30), ('L2', 'seminar', 15), ('L3', 'seminar', 10);

INSERT INTO enrollments (student_id, lesson_id)
VALUES ('S1', 'L1'), ('S1', 'L2'), ('S2', 'L1');

При включённых внешних ключах порядок важен: сначала студенты и занятия, потом записи, которые на них ссылаются

Если вставить строку с уже занятым ключом, запрос упадёт с IntegrityError
ON CONFLICT DO NOTHING в конце запроса меняет это поведение на тихий пропуск строки - именно так устроены add_student и add_lesson в примере ниже:

INSERT INTO students (student_id, name) VALUES ('S1', 'Алиса')
ON CONFLICT DO NOTHING;

SELECT - прочитать строки

SELECT lesson_id, capacity
FROM lessons
WHERE kind = 'seminar'
ORDER BY capacity DESC
LIMIT 1;
-- L2 | 15

Части запроса пишутся именно в таком порядке:

• SELECT - какие колонки вернуть: список через запятую или *, если нужны все колонки
• FROM - из какой таблицы брать строки
• WHERE - какие строки оставить
• ORDER BY - в каком порядке вернуть: по возрастанию по умолчанию (ASC), по убыванию - DESC
• LIMIT - сколько строк вернуть самое большее

Без ORDER BY порядок строк в результате не гарантирован - если он важен, сортировку нужно просить явно

Условие в WHERE собирается из таких кусочков:

• = - равно (знак один, не два), <> или != - не равно
• <, >, <=, >= - сравнение по величине
• AND, OR, NOT - соединить или отрицать условия, порядок задают скобки
• IN ('S2', 'S3') - значение входит в список
• LIKE 'Б%' - значение подходит под шаблон: % заменяет любое число символов, _ - один символ
• IS NULL и IS NOT NULL - проверка, есть ли значение

SELECT * FROM lessons WHERE capacity >= 15 AND kind = 'seminar';
-- L2 | seminar | 15

SELECT name FROM students WHERE name LIKE 'Б%' OR student_id IN ('S3') ORDER BY name;
-- Боря
-- Вера

NULL значит «значения нет» - это не пустая строка и не ноль
Сравнение с NULL через = никогда не бывает истинным, поэтому условие = NULL не вернёт ни одной строки, а отсутствие значения проверяют через IS NULL (пример - в разделе про LEFT JOIN ниже)

Агрегаты и GROUP BY

Агрегатные функции сворачивают много строк в одно значение: COUNT считает строки, SUM суммирует, а MIN, MAX и AVG находят минимум, максимум и среднее

SELECT COUNT(*), SUM(capacity), MAX(capacity) FROM lessons;
-- 3 | 55 | 30

COUNT(*) считает все строки, а COUNT(колонка) - только те, где в колонке не NULL
GROUP BY разбивает строки на группы с одинаковым значением колонки, и агрегаты считаются отдельно в каждой группе:

SELECT kind, COUNT(*) AS lessons_count, SUM(capacity) AS seats
FROM lessons
GROUP BY kind
ORDER BY kind;
-- lecture | 1 | 30
-- seminar | 2 | 25

AS задаёт псевдоним - короткое имя для колонки результата или для таблицы
Колонки в SELECT, которые не обёрнуты в агрегат, нужно перечислять и в GROUP BY: PostgreSQL иначе откажется выполнять запрос, а SQLite выполнит и вернёт значение из какой-то строки группы

WHERE отбирает строки до группировки и не может использовать агрегаты, а HAVING отбирает готовые группы после неё:

SELECT kind, COUNT(*) AS lessons_count
FROM lessons
GROUP BY kind
HAVING COUNT(*) > 1;
-- seminar | 2

JOIN - собрать данные из нескольких таблиц

SELECT s.name, e.lesson_id
FROM students AS s
JOIN enrollments AS e ON e.student_id = s.student_id
ORDER BY s.name, e.lesson_id;
-- Алиса | L1
-- Алиса | L2
-- Боря | L1

JOIN склеивает строки двух таблиц, для которых выполнено условие после ON - обычно «внешний ключ равен первичному ключу»
S3 (Вера) в результате нет: на занятия она не записана, и пары в enrollments у неё нет

Чтобы строки без пары не терялись, берут LEFT JOIN: все строки левой таблицы (той, что стоит после FROM) остаются, а колонки правой для них заполняются NULL:

SELECT l.lesson_id, l.capacity, COUNT(e.student_id) AS taken
FROM lessons AS l
LEFT JOIN enrollments AS e ON e.lesson_id = l.lesson_id
GROUP BY l.lesson_id, l.capacity
ORDER BY l.lesson_id;
-- L1 | 30 | 2
-- L2 | 15 | 1
-- L3 | 10 | 0

Занятие L3 без записей попало в результат с нулём: COUNT(e.student_id) не считает NULL
Тот же приём вместе с IS NULL находит строки, у которых пары нет совсем, - например, студентов без единой записи:

SELECT s.name
FROM students AS s
LEFT JOIN enrollments AS e ON e.student_id = s.student_id
WHERE e.student_id IS NULL;
-- Вера

UPDATE и DELETE - изменить и удалить

UPDATE lessons SET capacity = capacity + 5 WHERE lesson_id = 'L2';

DELETE FROM enrollments WHERE student_id = 'S1' AND lesson_id = 'L2';

UPDATE присваивает новые значения колонкам из SET в тех строках, которые подошли под WHERE, а в правой части можно использовать текущее значение колонки (capacity + 5)
DELETE удаляет подошедшие строки целиком
Без WHERE затрагиваются все строки таблицы, поэтому условие сначала проверяют обычным SELECT с тем же WHERE
Сколько строк изменил или удалил запрос, показывает cursor.rowcount: ноль означает, что ни одна строка не подошла

Параметризованные запросы и SQL-инъекции

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

name = "x' OR '1'='1"

# плохо: значение вклеено в текст запроса
conn.execute(f"SELECT * FROM users WHERE name = '{name}'")
# получился запрос: SELECT * FROM users WHERE name = 'x' OR '1'='1' - вернёт всех

# хорошо: значение передаётся отдельно от текста запроса
conn.execute("SELECT * FROM users WHERE name = ?", (name,))

Вместо значения в тексте запроса стоит плейсхолдер, а сами значения передаются вторым аргументом: драйвер подставляет их сам, и они никогда не читаются как SQL
sqlite3 поддерживает два стиля плейсхолдеров:

• ? - значения передаются кортежем или списком по порядку: ("S1", "Алиса")
• :name - значения передаются словарём по именам: {"student_id": "S1"}

Для одного значения нужен кортеж с запятой, (name,) - без запятой (name) это просто строка в скобках
Число значений должно совпадать с числом плейсхолдеров, иначе будет sqlite3.ProgrammingError
Плейсхолдеры подставляют только значения: имя таблицы или колонки через ? передать нельзя (SELECT * FROM ? - синтаксическая ошибка), а если оно приходит снаружи, его сверяют со списком допустимых имён

Много строк за один вызов вставляет executemany:

conn.executemany(
    "INSERT INTO students (student_id, name) VALUES (?, ?)",
    [("S1", "Алиса"), ("S2", "Боря")],
)

Номер, который база выдала новой строке, возвращает cursor.lastrowid: execute() отдаёт курсор, и сразу после INSERT в нём лежит номер вставленной строки
Тип у него int | None: пока вставки не было, значение равно None

cur = conn.execute("INSERT INTO notes (text) VALUES (?)", ("третья",))
print(cur.lastrowid)  # 3

Транзакции

Транзакция - группа операций, которая либо применяется целиком, либо не применяется совсем
Запись на занятие - как раз такой случай: нужно вставить запись и убедиться, что мест хватило, а если не хватило, вставка не должна остаться в базе

sqlite3 сам открывает транзакцию перед первым INSERT, UPDATE или DELETE (простой SELECT её не открывает) и держит открытой до commit() или rollback()
Пока транзакция не зафиксирована, изменения видны только своему соединению, а закрытие соединения без commit() их выбрасывает

Менеджер контекста соединения делает commit() на нормальном выходе и rollback() при исключении:

with conn:
    conn.execute(
        "INSERT INTO students (student_id, name) VALUES (?, ?)", ("S3", "Вера")
    )
    conn.execute(
        "INSERT INTO students (student_id, name) VALUES (?, ?)", ("S3", "Вася")
    )
# второй INSERT повторяет первичный ключ: IntegrityError откатит и первый INSERT

with conn не закрывает соединение: он только фиксирует или откатывает транзакцию
Закрывать соединение нужно отдельно - через conn.close() или contextlib.closing(conn)

SQLite пускает одного писателя за раз: пока одно соединение держит незафиксированную запись, другое соединение, которое тоже пытается писать, ждёт и по истечении таймаута (по умолчанию 5 секунд, параметр timeout у connect()) получает ошибку sqlite3.OperationalError: database is locked
Поэтому транзакции с записью держат короткими: открыли, записали, зафиксировали

В Python 3.12 у connect() появился параметр autocommit: по умолчанию остаётся описанное выше поведение, а autocommit=False документация называет рекомендуемым вариантом и обещает сделать его умолчанием в будущих версиях Python
В примерах ниже используется поведение по умолчанию

Ошибки БД и внешние ключи

Все исключения модуля наследуются от sqlite3.Error, чаще других встречаются три:

• sqlite3.IntegrityError - данные нарушают ограничение таблицы
• sqlite3.OperationalError - проблема на стороне базы: нет таблицы, синтаксическая ошибка в SQL, база заблокирована
• sqlite3.ProgrammingError - неверное использование API: не то число параметров, работа с закрытым соединением

Внутри IntegrityError причину различает атрибут sqlite_errorname (Python 3.11 и новее):

• SQLITE_CONSTRAINT_PRIMARYKEY - повтор первичного ключа
• SQLITE_CONSTRAINT_UNIQUE - повтор значения в колонке с UNIQUE
• SQLITE_CONSTRAINT_NOTNULL - NULL в колонке с NOT NULL
• SQLITE_CONSTRAINT_CHECK - не прошло условие CHECK
• SQLITE_CONSTRAINT_FOREIGNKEY - ссылка на несуществующую строку по внешнему ключу

Внешние ключи SQLite по умолчанию не проверяет - FOREIGN KEY в CREATE TABLE без этого остаётся просто описанием
Проверку включает PRAGMA foreign_keys = ON, и делать это нужно отдельно для каждого нового соединения, сразу после connect(): внутри уже открытой транзакции команда молча ничего не меняет

Ошибки базы полезно заворачивать в собственные исключения, как в статьях про исключения и про requests: вызывающему коду важно, что студент уже записан, а не какое именно ограничение сработало в SQLite

sqlite3.Row и датаклассы

По умолчанию строка - кортеж, и обращаться к колонкам приходится по номеру
conn.row_factory = sqlite3.Row превращает строки в объекты, к которым можно обращаться и по номеру, и по имени колонки, а row.keys() отдаёт список имён:

conn.row_factory = sqlite3.Row

row = conn.execute(
    "SELECT student_id, name FROM students WHERE student_id = ?", ("S1",)
).fetchone()
print(row["name"], row[0], row.keys())  # Алиса S1 ['student_id', 'name']

Такую строку можно развернуть в датакласс из статьи про ООП:

from dataclasses import dataclass


@dataclass(frozen=True)
class Student:
    student_id: str
    name: str


student = Student(**row)  # Student(student_id='S1', name='Алиса')

psycopg - то же самое для PostgreSQL

SQLite живёт внутри процесса, а PostgreSQL - отдельная программа-сервер: она слушает порт, к ней подключаются по сети, и она рассчитана на много клиентов, которые пишут одновременно
Драйвер для Python - psycopg (версия 3):

uv add "psycopg[binary]"

Вариант [binary] поставляется со своей копией библиотеки libpq, поэтому системные библиотеки и компилятор не нужны
Для продакшена документация предпочитает сборку из исходников с системной libpq (psycopg[c]): обновления системы тогда подхватываются сами

Чтобы запускать примеры из этого раздела, нужен локальный сервер PostgreSQL
Его установку и настройку разбирают на курсе по БД, а без сервера раздел можно просто прочитать: остальная статья работает на одном SQLite
Параметры подключения собираются в строку DSN (хост, порт, база, пользователь, пароль), значения в ней нужно заменить на свои
Пароль, записанный прямо в код, годится только для локальных экспериментов

import psycopg
from psycopg.rows import dict_row

DSN = "host=localhost port=5432 dbname=postgres user=postgres password=postgres"

with psycopg.connect(DSN, row_factory=dict_row) as conn:
    conn.execute(
        "INSERT INTO students (student_id, name) VALUES (%s, %s)", ("S1", "Алиса")
    )
    rows = conn.execute("SELECT student_id, name FROM students").fetchall()
# здесь commit, а соединение закрыто

Что отличается от sqlite3:

• плейсхолдеры - %s (или %(name)s для словаря) вместо ?
• with psycopg.connect(...) на выходе делает commit() (при исключении - rollback()) и закрывает соединение, а with conn в sqlite3 его не закрывает
• conn.execute() есть и у psycopg: он создаёт курсор и возвращает его, как в sqlite3
• dict_row играет роль sqlite3.Row, только строки приходят обычными словарями
• нет executescript: несколько команд в одном execute() разрешены только без параметров
• типы колонок строгие ('abc' в INTEGER - ошибка), а внешние ключи проверяются всегда, без PRAGMA
• у курсора нет lastrowid: номер новой строки приходит из самого запроса, INSERT ... RETURNING note_id
• ошибки приходят из psycopg.errors - UniqueViolation, ForeignKeyViolation и другие, они наследуются от psycopg.IntegrityError
• после ошибки внутри транзакции PostgreSQL отвергает все следующие команды (InFailedSqlTransaction), пока не будет rollback()

Для явной транзакции соединение открывают в режиме autocommit=True и оборачивают нужные операции в with conn.transaction(): на входе BEGIN, на нормальном выходе COMMIT, при исключении откат

Писателей в PostgreSQL много, поэтому приём «вставили, посчитали, откатили, если не влезло» без блокировки ломается: две параллельные транзакции вставляют по строке, каждая не видит чужую незафиксированную вставку, обе считают, что место ещё было, и обе фиксируются - записей становится больше, чем capacity
Лечит блокировка строки занятия: SELECT ... FOR UPDATE заставляет вторую транзакцию дождаться конца первой

Примеры

Продолжаем сквозную модель курсов и студентов: тот же Registrar, только теперь его состояние хранится в файле базы и переживает завершение скрипта

import sqlite3
from contextlib import closing

SCHEMA = """
CREATE TABLE IF NOT EXISTS students (
    student_id TEXT NOT NULL PRIMARY KEY,
    name       TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS lessons (
    lesson_id TEXT NOT NULL PRIMARY KEY,
    kind      TEXT NOT NULL CHECK (kind IN ('lecture', 'seminar')),
    capacity  INTEGER NOT NULL CHECK (capacity > 0)
);

CREATE TABLE IF NOT EXISTS enrollments (
    student_id TEXT NOT NULL REFERENCES students (student_id),
    lesson_id  TEXT NOT NULL REFERENCES lessons (lesson_id),
    PRIMARY KEY (student_id, lesson_id)
);
"""


class RegistrarError(Exception):
    """Базовая ошибка записи на занятия."""


class UnknownLessonError(RegistrarError):
    """Такого занятия нет."""


class UnknownStudentError(RegistrarError):
    """Такого студента нет."""


class AlreadyEnrolledError(RegistrarError):
    """Студент уже записан на это занятие."""


class LessonFullError(RegistrarError):
    """На занятии нет свободных мест."""


class RegistrarDB:
    def __init__(self, path: str) -> None:
        self.conn = sqlite3.connect(path)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA foreign_keys = ON")
        self.conn.executescript(SCHEMA)

    def close(self) -> None:
        self.conn.close()

    def add_student(self, student_id: str, name: str) -> None:
        with self.conn:
            self.conn.execute(
                "INSERT INTO students (student_id, name) VALUES (?, ?) "
                "ON CONFLICT DO NOTHING",
                (student_id, name),
            )

    def add_lesson(self, lesson_id: str, kind: str, capacity: int) -> None:
        with self.conn:
            self.conn.execute(
                "INSERT INTO lessons (lesson_id, kind, capacity) VALUES (?, ?, ?) "
                "ON CONFLICT DO NOTHING",
                (lesson_id, kind, capacity),
            )

    def enroll(self, student_id: str, lesson_id: str) -> None:
        with self.conn:
            lesson = self.conn.execute(
                "SELECT capacity FROM lessons WHERE lesson_id = ?", (lesson_id,)
            ).fetchone()
            if lesson is None:
                raise UnknownLessonError(lesson_id)

            try:
                self.conn.execute(
                    "INSERT INTO enrollments (student_id, lesson_id) VALUES (?, ?)",
                    (student_id, lesson_id),
                )
            except sqlite3.IntegrityError as e:
                if e.sqlite_errorname == "SQLITE_CONSTRAINT_PRIMARYKEY":
                    raise AlreadyEnrolledError(student_id, lesson_id) from e
                if e.sqlite_errorname == "SQLITE_CONSTRAINT_FOREIGNKEY":
                    raise UnknownStudentError(student_id) from e
                raise

            taken = self.conn.execute(
                "SELECT COUNT(*) FROM enrollments WHERE lesson_id = ?", (lesson_id,)
            ).fetchone()[0]
            if taken > lesson["capacity"]:
                raise LessonFullError(lesson_id)

    def lesson_load(self) -> list[sqlite3.Row]:
        return self.conn.execute(
            """
            SELECT l.lesson_id, l.capacity, COUNT(e.student_id) AS taken
            FROM lessons AS l
            LEFT JOIN enrollments AS e ON e.lesson_id = l.lesson_id
            GROUP BY l.lesson_id, l.capacity
            ORDER BY l.lesson_id
            """
        ).fetchall()


if __name__ == "__main__":
    with closing(RegistrarDB("registrar.db")) as registrar:
        registrar.add_student("S1", "Алиса")
        registrar.add_student("S2", "Боря")
        registrar.add_lesson("L1", "lecture", capacity=30)
        registrar.add_lesson("L2", "seminar", capacity=1)

        attempts = [
            ("S1", "L1"),
            ("S1", "L1"),
            ("S1", "L2"),
            ("S2", "L2"),
            ("S9", "L1"),
            ("S1", "L9"),
        ]
        for student_id, lesson_id in attempts:
            try:
                registrar.enroll(student_id, lesson_id)
                print(student_id, lesson_id, "записан")
            except RegistrarError as e:
                print(student_id, lesson_id, f"отказ: {type(e).__name__}")

        for row in registrar.lesson_load():
            print(row["lesson_id"], f"{row['taken']}/{row['capacity']}")

Первый запуск создаёт файл registrar.db и выводит:

S1 L1 записан
S1 L1 отказ: AlreadyEnrolledError
S1 L2 записан
S2 L2 отказ: LessonFullError
S9 L1 отказ: UnknownStudentError
S1 L9 отказ: UnknownLessonError
L1 1/30
L2 1/1

Второй запуск находит тот же файл, и записи первого запуска никуда не делись - в отличие от registrar из статьи про декораторы, который каждый запуск начинал заново:

S1 L1 отказ: AlreadyEnrolledError
S1 L1 отказ: AlreadyEnrolledError
S1 L2 отказ: AlreadyEnrolledError
S2 L2 отказ: LessonFullError
S9 L1 отказ: UnknownStudentError
S1 L9 отказ: UnknownLessonError
L1 1/30
L2 1/1

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

• вся операция идёт внутри with self.conn: любое исключение откатывает транзакцию целиком, включая уже сделанный INSERT
• сначала SELECT проверяет, что занятие существует, иначе UnknownLessonError
• INSERT записи сам проверяет остальное: повтор первичного ключа означает, что студент уже записан, а нарушение внешнего ключа - что такого студента нет (занятие уже проверено выше)
• после вставки COUNT(*) показывает, сколько записей на занятии; если их больше capacity, исключение LessonFullError откатывает только что вставленную запись
• такая проверка честная потому, что INSERT в SQLite уже занял блокировку записи: другие соединения не могут вставлять до конца транзакции, и между вставкой и подсчётом никто не вклинится

Схема SCHEMA из примера без изменений работает и в PostgreSQL, а в классе для psycopg меняются соединение, плейсхолдеры, перехват ошибок и блокировка строки занятия:

import psycopg
from psycopg.errors import ForeignKeyViolation, UniqueViolation
from psycopg.rows import dict_row

DSN = "host=localhost port=5432 dbname=postgres user=postgres password=postgres"


class RegistrarPG:
    def __init__(self, dsn: str) -> None:
        self.conn = psycopg.connect(dsn, autocommit=True, row_factory=dict_row)
        self.conn.execute(SCHEMA)

    def close(self) -> None:
        self.conn.close()

    def enroll(self, student_id: str, lesson_id: str) -> None:
        with self.conn.transaction():
            lesson = self.conn.execute(
                "SELECT capacity FROM lessons WHERE lesson_id = %s FOR UPDATE",
                (lesson_id,),
            ).fetchone()
            if lesson is None:
                raise UnknownLessonError(lesson_id)

            try:
                self.conn.execute(
                    "INSERT INTO enrollments (student_id, lesson_id) VALUES (%s, %s)",
                    (student_id, lesson_id),
                )
            except UniqueViolation as e:
                raise AlreadyEnrolledError(student_id, lesson_id) from e
            except ForeignKeyViolation as e:
                raise UnknownStudentError(student_id) from e

            taken = self.conn.execute(
                "SELECT COUNT(*) AS n FROM enrollments WHERE lesson_id = %s",
                (lesson_id,),
            ).fetchone()["n"]
            if taken > lesson["capacity"]:
                raise LessonFullError(lesson_id)

Остальные методы - те же запросы, только с %s вместо ?
FOR UPDATE в первом SELECT держит строку занятия заблокированной до конца транзакции, так что параллельные записи на одно занятие выстраиваются в очередь и места не перебираются

Автогрейдер-задача этой недели («БД и SQLite») строится вокруг той же модели курсов и студентов, но набор таблиц, ограничений и методов там свой - копировать код примера один в один не получится

Проверь себя

1. Чем WHERE отличается от HAVING?

WHERE отбирает строки до группировки и не может использовать агрегаты
HAVING отбирает готовые группы после GROUP BY, поэтому условие вроде COUNT(*) > 1 пишут именно в нём

2. Чем JOIN отличается от LEFT JOIN?

JOIN оставляет только те строки, для которых нашлась пара по условию ON
LEFT JOIN сохраняет и строки левой таблицы без пары, заполняя колонки правой значением NULL - поэтому занятие без записей пропадёт из результата JOIN, но останется в LEFT JOIN с COUNT(e.student_id), равным 0

3. Почему WHERE name = NULL не вернёт ни одной строки и как проверить отсутствие значения?

NULL значит «значения нет», и сравнение с ним через = никогда не бывает истинным
Правильно писать WHERE name IS NULL, а если нужны строки со значением - IS NOT NULL

4. Почему нельзя собирать SQL f-строкой с данными пользователя и как делать правильно?

Значение из f-строки становится частью текста запроса: ввод вроде x' OR '1'='1 меняет смысл условия (это и есть SQL-инъекция)
Правильно - плейсхолдеры (? или :name) и значения вторым аргументом execute: драйвер передаёт их отдельно от SQL, и они не могут стать его частью

5. Можно ли передать имя таблицы через ??

Нет: плейсхолдеры подставляют только значения, SELECT * FROM ? - синтаксическая ошибка
Имя таблицы или колонки, пришедшее снаружи, сверяют со списком допустимых имён и только потом вставляют в текст запроса

6. Что делает with conn: в sqlite3 и чего он не делает?

Фиксирует транзакцию (commit) на нормальном выходе и откатывает (rollback) при исключении
Соединение при этом остаётся открытым - закрывать его нужно отдельно, через conn.close() или contextlib.closing

7. Что окажется в базе, если сделать INSERT, не вызвать commit() и закрыть соединение?

Ничего: незафиксированная транзакция пропадает вместе с соединением
Поэтому запись оборачивают в with conn: или явно зовут commit()

8. Почему без PRAGMA foreign_keys = ON запись для несуществующего студента вставляется, а с ним падает, и когда этот PRAGMA выполнять?

SQLite по умолчанию не проверяет внешние ключи, FOREIGN KEY в CREATE TABLE остаётся только описанием
PRAGMA foreign_keys = ON включает проверку, но лишь для того соединения, где его выполнили, и не действует внутри открытой транзакции - поэтому его вызывают сразу после каждого connect()
С проверкой вставка падает с sqlite3.IntegrityError (SQLITE_CONSTRAINT_FOREIGNKEY)

9. Чем IntegrityError отличается от OperationalError и как внутри IntegrityError отличить повтор ключа от неверной ссылки?

IntegrityError - данные нарушают ограничение таблицы (ключ, UNIQUE, NOT NULL, CHECK, внешний ключ), OperationalError - проблема на стороне базы (нет таблицы, синтаксис, блокировка)
Причину внутри IntegrityError показывает e.sqlite_errorname: SQLITE_CONSTRAINT_PRIMARYKEY для повтора ключа и SQLITE_CONSTRAINT_FOREIGNKEY для неверной ссылки

10. Что в psycopg ведёт себя иначе, чем в sqlite3?

Плейсхолдеры %s вместо ?
with psycopg.connect(...) не только фиксирует или откатывает, но и закрывает соединение
После ошибки в транзакции PostgreSQL отвергает команды до rollback()
Типы колонок и внешние ключи проверяются всегда, а ошибки приходят из psycopg.errors (UniqueViolation, ForeignKeyViolation) и наследуются от psycopg.IntegrityError

11. Зачем в версии для PostgreSQL нужен FOR UPDATE, а в версии для SQLite хватает вставки и подсчёта?

В SQLite писатель один: после INSERT база заблокирована на запись для остальных до конца транзакции, и подсчёт получается честным
В PostgreSQL транзакции пишут параллельно и не видят чужих незафиксированных вставок - без блокировки две записи на последнее место могли бы пройти обе
FOR UPDATE блокирует строку занятия, и вторая транзакция ждёт, пока первая завершится

Итог / чек-лист

• реляционная модель - таблицы, первичный и внешний ключи, связь «многие ко многим» через таблицу-связку
• CREATE TABLE описывает колонки, типы и ограничения (PRIMARY KEY, NOT NULL, UNIQUE, DEFAULT, CHECK, REFERENCES), а ON CONFLICT DO NOTHING превращает вставку дубликата в тихий пропуск
• числовой ключ INTEGER PRIMARY KEY нумеруется базой сам, а номер новой строки отдаёт cursor.lastrowid (в PostgreSQL - GENERATED ... AS IDENTITY и RETURNING)
• порядок частей SELECT фиксирован - FROM, WHERE, GROUP BY, HAVING, ORDER BY, LIMIT, а без ORDER BY порядок строк не гарантирован
• WHERE отбирает строки до группировки, HAVING - группы после неё, а отсутствие значения проверяют через IS NULL, не через = NULL
• JOIN оставляет только строки с парой, LEFT JOIN сохраняет и строки без пары, подставляя NULL
• UPDATE и DELETE без WHERE затрагивают всю таблицу, число затронутых строк - в cursor.rowcount
• sqlite3 - connect, execute, fetchone/fetchall, commit, close, а ":memory:" для экспериментов
• значения попадают в запрос только через плейсхолдеры (?, :name), никогда через f-строки, а имена таблиц и колонок плейсхолдерами не передаются
• транзакция открывается неявно перед INSERT/UPDATE/DELETE, with conn фиксирует или откатывает, но не закрывает соединение
• PRAGMA foreign_keys = ON нужен после каждого connect(), иначе внешние ключи не проверяются
• ошибки IntegrityError, OperationalError, ProgrammingError различаются по смыслу, причину ограничения даёт sqlite_errorname, а наружу ошибки БД выходят собственными исключениями
• sqlite3.Row даёт доступ к колонкам по имени и разворачивается в датакласс
• psycopg - %s, with закрывает соединение, типы строгие, а проверки вместимости при параллельных писателях требуют FOR UPDATE

Дополнительно

• sqlite3 - DB-API 2.0 interface for SQLite databases
• SQLite - Datatypes
• SQLite - STRICT Tables
• SQLite - Foreign Key Support
• Psycopg 3 - Basic module usage
• Psycopg 3 - Transactions management
• Psycopg 3 - Installation
• PostgreSQL - Explicit Locking
• xkcd 327 - Exploits of a Mom