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

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

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

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

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

Python не «владеет» базой данных: он подключается к уже запущенной СУБД (или к файлу SQLite) через драйвер — библиотеку, реализующую общий контракт DB-API 2.0 (PEP 249). Поверх драйвера часто ставят ORM (объектно-реляционное отображение), миграции схемы и валидацию моделей — но понимание «сырого» цикла connect → cursor → execute → fetch → commit остаётся обязательным: без него сложно отлаживать утечки соединений, блокировки и медленные запросы.

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

Загрузка демо Python + БД…

Демо показывает стек 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()

Параметризованные запросы (? или %s / %(name)s в зависимости от драйвера) — обязательная практика: конкатенация строк с пользовательским вводом открывает SQL-инъекции.

Атрибуты модуля драйвера по 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"

Пул соединений (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"],
)

---

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

СУБД	Встроенный / синхронный драйвер	Асинхронный	Типичное применение
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):

import sqlite3

conn = sqlite3.connect("example.db")
conn.execute("PRAGMA foreign_keys = ON")
cur = conn.cursor()
cur.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT UNIQUE NOT NULL,
        email TEXT NOT NULL
    )
""")
cur.execute("INSERT INTO users (username, email) VALUES (?, ?)", ("bob", "bob@example.com"))
conn.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())

Асинхронный доступ (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())

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

---

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

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

from datetime import datetime
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

engine = create_engine("sqlite:///example.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

with Session() as session:
    session.add(User(username="charlie", email="charlie@example.com"))
    session.commit()
    for u in session.query(User).filter(User.username == "charlie"):
        print(u.id, u.email)

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

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

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

---

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

Инструмент	Назначение
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 или драйвер

---

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

Транзакция — атомарная единица: 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()

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()

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

---

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

• Медленные запросы: логи PostgreSQL log_min_duration_statement, EXPLAIN (ANALYZE, BUFFERS) — см. оптимизацию SQL.
• Активные сессии: 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

Пул: следите за 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)

Для больших выборок — потоковая выборка (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())

Продакшен-данные на 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 и миграции по мере роста проекта; резервирование и мониторинг закладывайте до первого продакшен-инцидента, а не после.

---