Работа с базами данных в Python

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику Архитектору

См. также: FastAPI и база данных · Работа с файлами, сетью и внешними API · Django · раздел SQL · управление СУБД

---

Работа с базами данных

Python подключается к уже запущенной СУБД (PostgreSQL, MySQL…) или к файлу SQLite на диске. Сам по себе Python базу не "поднимает" — нужен сервер PostgreSQL или файл app.db.

Главная идея этой главы: сначала понимать низкоуровневый цикл DB-API, потом использовать ORM как ускоритель разработки. Тогда даже при сложных запросах и инцидентах в продакшене вы сохраняете контроль над SQL и транзакциями.

---

С чего начать новичку

Рекомендуемый порядок в этой главе:

1. Прочитать словарь ниже.
2. Пройти пример sqlite3 + DB-API (connect → cursor → execute).
3. Понять транзакцию commit / rollback.
4. При веб-разработке перейти к ORM (SQLAlchemy) — FastAPI и БД, Flask.
5. Перед продакшеном — миграции, бэкапы, мониторинг.

Драйвер — библиотека, которая говорит с конкретной СУБД (psycopg2, встроенный sqlite3). Все следуют DB-API 2.0 (PEP 249): connect, cursor, execute, fetchone.

ORM (SQLAlchemy, Django ORM) строит SQL из классов Python. Цикл connect → cursor → execute → fetch → commit нужен для отладки утечек соединений, блокировок и медленных запросов, когда ORM генерирует неожиданный SQL.

---

Словарь

Термин	Простыми словами
СУБД	Программа с таблицами: PostgreSQL, MySQL, SQLite
Строка (row)	Одна запись в таблице
Курсор	Объект для отправки SQL и чтения результата
Параметризованный запрос	WHERE id = ? + значения — защита от SQL-инъекций
Транзакция	Группа изменений: все commit или все rollback
DSN / URL	Строка подключения: хост, порт, БД, логин
Пул соединений	Переиспользование открытых сессий к серверу БД

---

Интерактивная лаборатория

Play ITЗагрузка интерактивного демо…

Демо показывает стек DB-API (соединение, курсор, SQL), типовой CRUD, слои доступа (драйвер → ORM → фреймворк) и жизненный цикл транзакции. В главе Django тот же компонент доступен в режиме variant="django" (MTV и ORM Django).

---

DB-API 2.0 — единый контракт

Спецификация задаёт минимальный интерфейс:

Объект	Назначение
Connection	Сессия с СУБД; commit(), rollback(), close()
Cursor	Выполнение SQL; execute(), executemany(), fetchone(), fetchall()
Исключения	Иерархия Error, DatabaseError, IntegrityError и др.

Типичный синхронный цикл:

import sqlite3

with sqlite3.connect("app.db") as conn:
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO users (username, email) VALUES (?, ?)",
        ("alice", "alice@example.com"),
    )
    cur.execute("SELECT id, username FROM users WHERE username = ?", ("alice",))
    row = cur.fetchone()

Разбор фрагмента:

• sqlite3.connect("app.db") открывает или создаёт SQLite-базу в файле.
• Контекстный менеджер with ... as conn автоматически закрывает соединение.
• conn.cursor() создаёт курсор для выполнения SQL-команд.
• execute(..., (...)) использует параметризованные запросы и защищает от SQL-инъекций.
• fetchone() возвращает первую найденную строку результата.

Практически это означает, что при смене СУБД чаще меняется драйвер и строка подключения, а архитектурный каркас кода остаётся знакомым. Именно это делает DB-API важным "общим языком" между проектами на разных базах.

---

Разбор построчно

Строка	Смысл
sqlite3.connect("app.db")	Открыть/создать файл БД
with ... as conn	Закрыть соединение при выходе
conn.cursor()	Курсор для SQL
execute(SQL, кортеж)	Запрос; ? — места для параметров
fetchone()	Одна строка или None

Параметризованные запросы (? в SQLite, %s в psycopg2) — обязательная практика.

Опасный антипример:

# ОПАСНО — пользователь может подставить свой SQL
username = request.args["name"]
cur.execute(f"SELECT * FROM users WHERE username = '{username}'")

С параметрами драйвер передаёт значение как данные, а не как часть команды.

Атрибуты модуля драйвера по PEP 249: apilevel (например '2.0'), threadsafety, paramstyle (qmark, pyformat, named).

---

Как устроено подключение

1. Строка подключения (DSN) — URL или набор параметров — хост, порт, имя БД, пользователь, SSL.
2. Handshake — TCP (или локальный сокет), аутентификация, выбор кодировки и схемы (search_path в PostgreSQL).
3. Сессия — одно соединение обслуживает последовательность транзакций до close().

Примеры DSN:

# SQLite — файл на диске
"sqlite:///./local.db"

# PostgreSQL (SQLAlchemy / многие ORM)
"postgresql+psycopg2://user:pass@localhost:5432/mydb"

# MySQL
"mysql+pymysql://user:pass@localhost:3306/mydb"

Фрагмент показывает форматы строк подключения для разных СУБД:

• префикс (sqlite, postgresql+psycopg2, mysql+pymysql) определяет диалект и драйвер;
• после :// идут параметры доступа: пользователь, пароль, хост, порт и имя базы;
• одна и та же идея DSN используется в ORM, скриптах миграции и сервисах.

Пул соединений (SQLAlchemy create_engine(..., pool_size=5), psycopg2.pool) переиспользует открытые сессии в веб-приложениях и воркерах — иначе каждый HTTP-запрос открывал бы новое TCP-соединение к СУБД.

Переменные окружения для секретов:

import os
import psycopg2

conn = psycopg2.connect(
    host=os.environ["PGHOST"],
    dbname=os.environ["PGDATABASE"],
    user=os.environ["PGUSER"],
    password=os.environ["PGPASSWORD"],
)

Этот код берёт параметры подключения из переменных окружения:

• os.environ[...] читает секреты и конфигурацию без хардкода в репозитории;
• psycopg2.connect(...) открывает сессию к PostgreSQL с переданными значениями;
• подход удобен для локальной разработки, CI/CD и контейнерных окружений.

---

СУБД, драйверы и когда что выбирать

СУБД	Встроенный / синхронный драйвер	Асинхронный	Типичное применение
SQLite	sqlite3 (stdlib)	aiosqlite	Прототипы, тесты, встраиваемые приложения
PostgreSQL	psycopg2, psycopg (v3)	asyncpg	Продакшен, JSONB, расширения
MySQL / MariaDB	mysql-connector-python, PyMySQL	aiomysql	Веб-стеки LAMP, совместимость
MS SQL Server	pyodbc, pymssql	ограниченно	Корпоративные Windows-среды
Oracle	oracledb (бывш. cx_Oracle)	—	Enterprise
MongoDB	pymongo	motor	Документы, гибкая схема
Redis	redis	redis.asyncio	Кэш, очереди, не полноценная SQL-БД

Серверные СУБД дают конкурентную запись, репликацию и администрирование; SQLite — один файл, блокировка на уровне файла при интенсивной записи.

SQLite (stdlib):

Код ITЗагрузка примера кода…

Здесь показан минимальный CRUD-цикл с SQLite:

• connect("example.db") открывает файл базы;
• PRAGMA foreign_keys = ON включает контроль внешних ключей;
• CREATE TABLE IF NOT EXISTS создаёт таблицу при первом запуске;
• INSERT ... VALUES (?, ?) добавляет запись параметризованно;
• commit() фиксирует транзакцию на диск.

PostgreSQL (psycopg2):

import psycopg2

with psycopg2.connect(
    host="localhost", database="mydb", user="user", password="pass"
) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT version();")
        print(cur.fetchone())

Фрагмент демонстрирует безопасную работу через контекстные менеджеры:

• внешний with psycopg2.connect(...) управляет соединением и транзакцией;
• внутренний with conn.cursor() as cur создаёт и закрывает курсор автоматически;
• SELECT version(); проверяет доступность сервера и отвечает его версией.

Асинхронный доступ (asyncpg + asyncio):

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect(
        host="localhost", database="mydb", user="user", password="pass"
    )
    rows = await conn.fetch("SELECT id, username FROM users LIMIT 10")
    await conn.close()

asyncio.run(main())

Пример показывает неблокирующий доступ к БД в event loop:

• await asyncpg.connect(...) открывает асинхронное соединение;
• await conn.fetch(...) получает набор строк без блокировки потока;
• await conn.close() явно закрывает ресурсы;
• asyncio.run(main()) запускает корутину как точку входа.

В FastAPI и asyncio-приложениях асинхронные драйверы не блокируют event loop на ожидании сети.

---

ORM и уровни абстракции

SQLAlchemy — де-факто стандарт: уровень Core (SQL-выражения) и ORM (классы ↔ таблицы).

Код ITЗагрузка примера кода…

Этот блок строит полный базовый поток SQLAlchemy ORM:

• declarative_base() создаёт базовый класс для моделей;
• класс User задаёт таблицу, колонки и ограничения;
• create_engine(...) настраивает подключение, create_all(...) поднимает схему;
• Session = sessionmaker(...) создаёт фабрику сессий;
• в with Session() добавляется объект, фиксируется commit(), затем выполняется выборка.

Django ORM — часть фреймворка; миграции makemigrations / migrate, QuerySet API. Подробнее — в Django.

Другие — Peewee, Tortoise ORM (async), SQLModel (Pydantic + SQLAlchemy).

Правило: ORM ускоряет CRUD, но сложные отчёты и тонкая оптимизация часто требуют явного SQL или session.execute(text("...")).

---

Когда переходить от "чистого SQL" к ORM

Рабочий ориентир:

• если в проекте 2–4 таблицы и немного запросов, прямой SQL через DB-API полностью достаточен;
• если сущностей становится много и появляется бизнес-логика вокруг связей, ORM обычно ускоряет разработку;
• если команда смешанная по опыту, ORM снижает порог входа, а критичные запросы можно оставить на явном SQL;
• если много аналитических выборок и специфики СУБД, архитектура "ORM + SQL для тяжёлых кейсов" даёт лучший баланс.

Этот переход удобно планировать вместе с вопросами производительности из разделов SQL — оптимизация и индексы в SQL.

---

Мини-чеклист перед продакшеном

• Настроены миграции и процедура отката (downgrade) для критичных изменений схемы.
• Включены бэкапы и проверка восстановления на отдельном стенде.
• В коде только параметризованные запросы.
• Есть таймауты подключения и запросов.
• Мониторятся медленные запросы и насыщение пула соединений.
• Учтены права доступа: отдельные роли для приложения, миграций и администрирования.

Такой чеклист особенно полезен перед релизом API на FastAPI или Django, где нагрузка на базу резко растёт после публикации.

---

Миграции схемы и валидация данных

Инструмент	Назначение
Alembic	Миграции для SQLAlchemy (версии, upgrade/downgrade)
Django migrations	Автогенерация из models.py
Pydantic	Валидация DTO до/после БД (не заменяет CHECK в СУБД)
jsonschema	Валидация JSON-полей и конфигов

Пример Alembic (концептуально) — ревизии в versions/, команды alembic upgrade head, alembic revision --autogenerate -m "add column".

Валидация на границе приложения:

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    username: str
    email: EmailStr

payload = UserCreate(username="dave", email="dave@example.com")
# далее — INSERT через ORM или драйвер

Фрагмент вводит валидацию входных данных до записи в БД:

• BaseModel задаёт схему данных в коде;
• EmailStr проверяет формат email на этапе создания объекта;
• создание UserCreate(...) сразу выявляет невалидный payload и предотвращает "грязные" данные в таблицах.

---

Транзакции и целостность

Транзакция — атомарная единица: commit фиксирует все изменения, rollback отменяет.

import sqlite3

conn = sqlite3.connect("example.db")
try:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise
finally:
    conn.close()

Разбор фрагмента:

• Два UPDATE образуют одну транзакцию перевода между счетами.
• commit() фиксирует изменения только если обе операции прошли успешно.
• В except вызывается rollback(), чтобы сохранить целостность данных при ошибке.
• finally гарантирует закрытие соединения независимо от исхода.

SQLAlchemy: with session.begin(): — автоматический commit/rollback.

---

Резервное копирование и восстановление

Подход зависит от СУБД (см. резервное копирование SQL):

• PostgreSQL — pg_dump, pg_restore, логическая/физическая репликация.
• MySQL: mysqldump, binlog.
• SQLite: копирование файла .db при отсутствии активной записи или через backup API:

import sqlite3

src = sqlite3.connect("production.db")
dst = sqlite3.connect("backup.db")
src.backup(dst)
src.close()
dst.close()

Этот код делает резервную копию SQLite на уровне движка:

• открываются два соединения: источник production.db и приёмник backup.db;
• src.backup(dst) переносит текущее состояние базы в файл бэкапа;
• закрытие обоих соединений завершает операцию корректно.

Из Python часто оркестрируют вызов утилит через subprocess.run с проверкой кода возврата и ротацией архивов (shutil, gzip).

---

Мониторинг и диагностика

• Медленные запросы: логи PostgreSQL log_min_duration_statement, — см. оптимизацию SQL.

EXPLAIN (ANALYZE, BUFFERS)

Команда используется для анализа плана выполнения SQL-запроса:

• EXPLAIN показывает, какие шаги планирует оптимизатор;

• ANALYZE добавляет фактические времена и количество строк;

• BUFFERS раскрывает работу с кэшем и чтением страниц с диска.

• Активные сессии: pg_stat_activity, SHOW PROCESSLIST (MySQL).

• Из Python: psycopg2 + запрос к представлениям статистики; Prometheus-экспортеры; APM (OpenTelemetry, Sentry с трассировкой SQL).

cur.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", (email,))
print(cur.fetchall())  # SQLite

Здесь диагностика делается прямо из Python-кода:

• EXPLAIN QUERY PLAN ... просит SQLite показать план выбранного запроса;
• параметр (email,) сохраняет безопасность и повторяемость вызова;
• fetchall() возвращает шаги плана, которые помогают понять, используется ли индекс.

Пул: следите за pool.overflow, временем ожидания соединения и числом "зависших" транзакций без commit.

---

Экспорт, автоматизация запросов и отчёты

import csv
import sqlite3

conn = sqlite3.connect("example.db")
rows = conn.execute("SELECT id, username, email FROM users").fetchall()

with open("users_export.csv", "w", encoding="utf-8", newline="") as f:
    w = csv.writer(f)
    w.writerow(["id", "username", "email"])
    w.writerows(rows)

Фрагмент экспортирует выборку в CSV-файл:

• данные читаются из БД в rows через SELECT;
• csv.writer(...) формирует табличный текстовый формат;
• writerow(...) записывает заголовок колонок;
• writerows(rows) добавляет все строки результата в файл.

Для больших выборок — потоковая выборка (cursor.fetchmany(1000)), запись в Parquet (pyarrow), выгрузка в S3 (boto3). Планировщик — cron, Celery, Airflow.

---

Генерация тестовых данных и анонимизация

• Faker — правдоподобные имена, адреса, email для стендов.
• Анонимизация: хеширование идентификаторов (SHA-256 + соль), маскирование (user@example.com → u***@example.com), замена на синтетику; для GDPR — отдельные политики хранения.

from faker import Faker

fake = Faker("ru_RU")
for _ in range(5):
    print(fake.name(), fake.email())

Пример создаёт демонстрационные персональные данные:

• Faker("ru_RU") включает русскую локаль генерации;
• в цикле формируются разные имя и email для тестов и стендов;
• такие данные подходят для разработки без использования реальных персональных данных.

Продакшен-данные на dev-копиях без анонимизации — распространённая причина утечек.

---

Оптимизация структуры и доступа

• Нормализация / денормализация по нагрузке (см. нормализацию).
• Индексы под реальные фильтры и JOIN.
• N+1 в ORM — joinedload, selectinload в SQLAlchemy; select_related / prefetch_related в Django.
• Партиционирование, материализованные представления — на уровне СУБД.

---

NoSQL и гибридные сценарии

Redis — кэш сессий, rate limit, не замена реляционной БД для отчётов.

MongoDB — документы; схема гибкая, но индексы и агрегации всё равно проектируют осознанно.

Частый паттерн: PostgreSQL как источник истины + Redis как кэш + очередь (RabbitMQ, Kafka) для асинхронной обработки.

---

Выбор стека — краткая схема

Итог: начинайте с DB-API и понимания транзакций; добавляйте ORM и миграции по мере роста проекта; резервирование и мониторинг закладывайте до первого продакшен-инцидента, а не после.

---

Практикум по прикладным библиотекам для data-layer

SQLAlchemy и Alembic в одном цикле

Рабочий минимум:

1. Описать модели SQLAlchemy.
2. Создать ревизию alembic revision --autogenerate -m "...".
3. Применить миграции alembic upgrade head.
4. В API-слое валидировать вход через Pydantic.

Разбор этапов:

• Модели SQLAlchemy определяют структуру таблиц и связи между сущностями.
• alembic revision --autogenerate сравнивает модели с текущей схемой и готовит скрипт миграции.
• alembic upgrade head применяет все новые ревизии до актуального состояния.
• Pydantic на входе API отсекает невалидные данные до попадания в транзакцию БД.

Faker и factory_boy для тестовых данных

from faker import Faker

import factory

fake = Faker("ru_RU")

class UserFactory(factory.DictFactory):
    username = factory.LazyFunction(lambda: fake.user_name())
    email = factory.LazyFunction(lambda: fake.email())

Разбор фрагмента:

• Faker("ru_RU") генерирует реалистичные данные в русской локали.
• factory.DictFactory создаёт словари вместо ORM-объектов.
• LazyFunction(...) вызывает генератор при каждом создании записи, поэтому значения не повторяются.

Pympler для анализа памяти

from pympler import asizeof

rows = [{"id": i, "name": f"user-{i}"} for i in range(100_000)]
print(asizeof.asizeof(rows))  # байты в памяти

Разбор фрагмента:

• Создаётся крупная коллекция, имитирующая данные после выборки.
• asizeof.asizeof(rows) считает фактический объём объекта вместе с вложенными структурами.
• Такой замер помогает обнаруживать перепотребление памяти в ETL/репортах.

py-spy для профилирования запросных воркеров

Профилирование запускают вне кода:

py-spy top --pid <PID_ПРОЦЕССА>
py-spy record -o profile.svg --pid <PID_ПРОЦЕССА>

Разбор команд:

• py-spy top показывает "горячие" функции процесса в реальном времени.
• py-spy record записывает профиль и формирует flame graph в profile.svg.
• Это позволяет увидеть узкие места CPU без изменения исходного кода приложения.

---

Основа по протоколу

Базовый разбор HTTP и HTTPS находится в отдельной статье — HTTP как основа веб-интеграций.