Как взаимодействовать с реляционными базами данных?

Работа с реляционными базами данных в Python

В Python существует несколько подходов для взаимодействия с БД. Выбор зависит от сложности проекта и требований.

1. DB-API (Низкоуровневый доступ)

Стандартный интерфейс для взаимодействия с БД. Каждая БД имеет свой драйвер:

# PostgreSQL
import psycopg2

connection = psycopg2.connect(
    dbname="mydb",
    user="postgres",
    password="secret",
    host="localhost",
    port=5432
)

cursor = connection.cursor()

# SELECT
cursor.execute("SELECT * FROM users WHERE age > %s", (18,))
users = cursor.fetchall()
for user in users:
    print(user)

# INSERT
cursor.execute(
    "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
    ("John", "john@example.com", 25)
)
connection.commit()

# UPDATE
cursor.execute("UPDATE users SET age = %s WHERE name = %s", (26, "John"))
connection.commit()

# DELETE
cursor.execute("DELETE FROM users WHERE age < %s", (18,))
connection.commit()

cursor.close()
connection.close()

Другие драйверы БД

# MySQL
import mysql.connector
connection = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password",
    database="mydb"
)

# SQLite (встроенная БД)
import sqlite3
connection = sqlite3.connect("database.db")

# SQL Server
import pyodbc
connection = pyodbc.connect(
    "Driver={ODBC Driver 17 for SQL Server};Server=server;Database=db;UID=user;PWD=pass"
)

2. SQLAlchemy ORM (Высокоуровневый подход)

Объектно-реляционное отображение. Абстрагирует конкретную БД.

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

# Создание engine
engine = create_engine("postgresql://user:password@localhost/mydb")
Base = declarative_base()

# Определение модели
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)

# Создание таблиц
Base.metadata.create_all(engine)

# Создание сессии
Session = sessionmaker(bind=engine)
session = Session()

# CREATE
new_user = User(name="John", email="john@example.com", age=25)
session.add(new_user)
session.commit()

# READ
users = session.query(User).filter(User.age > 18).all()
for user in users:
    print(user.name, user.email)

# UPDATE
user = session.query(User).filter_by(name="John").first()
if user:
    user.age = 26
    session.commit()

# DELETE
user = session.query(User).filter_by(name="John").first()
if user:
    session.delete(user)
    session.commit()

session.close()

3. SQLAlchemy Core (Средний уровень)

Чистый SQL с удобством Python:

from sqlalchemy import create_engine, select, insert, update, delete, Table, Column, Integer, String, MetaData

engine = create_engine("postgresql://user:password@localhost/mydb")
metadata = MetaData()

users_table = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("email", String(100)),
    Column("age", Integer)
)

with engine.connect() as conn:
    # SELECT
    stmt = select(users_table).where(users_table.c.age > 18)
    result = conn.execute(stmt)
    for row in result:
        print(row)
    
    # INSERT
    stmt = insert(users_table).values(name="John", email="john@example.com", age=25)
    conn.execute(stmt)
    conn.commit()
    
    # UPDATE
    stmt = update(users_table).where(users_table.c.name == "John").values(age=26)
    conn.execute(stmt)
    conn.commit()
    
    # DELETE
    stmt = delete(users_table).where(users_table.c.age < 18)
    conn.execute(stmt)
    conn.commit()

4. Работа с миграциями (Alembic)

Управление изменениями схемы БД:

# Инициализация
alembic init migrations

# Создание миграции
alembic revision --autogenerate -m "Add age column to users"

# Применение миграции
alembic upgrade head

# Откат миграции
alembic downgrade -1

# Пример миграции (migrations/versions/001_initial.py)
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(50), nullable=False),
        sa.Column("email", sa.String(100), nullable=True),
        sa.PrimaryKeyConstraint("id")
    )

def downgrade():
    op.drop_table("users")

5. Управление соединениями (Connection Pooling)

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "postgresql://user:password@localhost/mydb",
    poolclass=QueuePool,
    pool_size=5,          # Минимум соединений в пуле
    max_overflow=10,      # Максимум дополнительных соединений
    pool_recycle=3600,    # Переиспользовать соединение через 1 час
    pool_pre_ping=True    # Проверять соединение перед использованием
)

6. Контекстный менеджер для безопасной работы

from contextlib import contextmanager

@contextmanager
def get_db_connection(connection_string):
    """Безопасное управление соединением"""
    connection = None
    try:
        connection = psycopg2.connect(connection_string)
        yield connection
    except Exception as e:
        print(f"Database error: {e}")
        if connection:
            connection.rollback()
    finally:
        if connection:
            connection.close()

# Использование
with get_db_connection("dbname=mydb user=postgres") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    users = cursor.fetchall()

7. Транзакции и откаты

from sqlalchemy.orm import Session

session = Session(engine)

try:
    # Несколько операций в одной транзакции
    user1 = User(name="John", email="john@example.com", age=25)
    user2 = User(name="Jane", email="jane@example.com", age=28)
    
    session.add(user1)
    session.add(user2)
    session.commit()  # Всё успешно
    
except Exception as e:
    session.rollback()  # Откат всех изменений
    print(f"Transaction failed: {e}")
finally:
    session.close()

8. Защита от SQL Injection

Всегда используй параметризованные запросы:

# ❌ ПЛОХО - уязвимо для SQL Injection
query = f"SELECT * FROM users WHERE email = {user_email}"
cursor.execute(query)

# ✅ ХОРОШО - защищено
cursor.execute("SELECT * FROM users WHERE email = %s", (user_email,))

# SQLAlchemy защищает автоматически
user = session.query(User).filter_by(email=user_email).first()

9. Оптимизация запросов

from sqlalchemy import joinedload

# N+1 проблема
users = session.query(User).all()
for user in users:
    print(user.posts)  # Дополнительный запрос для каждого пользователя

# Решение: eager loading
users = session.query(User).options(joinedload(User.posts)).all()

# Batch loading
from sqlalchemy.orm import contains_eager
users = session.query(User).join(Post).options(contains_eager(User.posts)).all()

10. Лучшие практики

• Используй ORM для бизнес-логики
• Используй Raw SQL только для сложных/специфических запросов
• Всегда управляй соединениями (используй context managers)
• Защищай от SQL Injection (параметризованные запросы)
• Оптимизируй N+1 проблемы (eager loading)
• Настраивай connection pooling для production
• Тестируй все БД операции

Выбор подхода зависит от сложности проекта, но в большинстве случаев SQLAlchemy ORM — лучший выбор для Python приложений.