Как работал с базами данных?

Как я работал с базами данных

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

Опыт с PostgreSQL

Я активно работал с PostgreSQL в нескольких долгосрочных проектах. PostgreSQL — это мощная реляционная БД, и я использовал её возможности по максимуму:

import psycopg2
from psycopg2.extras import RealDictCursor

# Прямая работа с курсором
connection = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="postgres",
    password="password"
)

cursor = connection.cursor(RealDictCursor)
cursor.execute(
    "SELECT id, name, email FROM users WHERE created_at > %s",
    ("2024-01-01",)  # Параметризованные запросы — защита от SQL injection
)

users = cursor.fetchall()
connection.close()

Я работал с:

• Транзакциями и ACID гарантиями для критичных операций
• Индексами для оптимизации запросов (B-tree, GIN, GIST)
• Представлениями (views) для абстрактизации сложной логики
• JSON типами для гибкого хранения данных
• Full-text search для поиска по тексту

Работа с ORM: SQLAlchemy

На большинстве проектов я использовал SQLAlchemy для более продуктивной разработки:

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

Base = declarative_base()

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

engine = create_engine('postgresql://user:password@localhost/mydb')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# CRUD операции
user = User(name="John", email="john@example.com")
session.add(user)
session.commit()

# Запросы
users = session.query(User).filter(User.name.like('%John%')).all()
session.close()

Оптимизация N+1 проблемы

Это была частая проблема в моей работе. Я научился решать её несколькими способами:

from sqlalchemy.orm import joinedload, selectinload

# Проблема: N+1 запросы
users = session.query(User).all()
for user in users:
    print(user.posts)  # Каждый раз — новый запрос!

# Решение 1: joinedload (LEFT JOIN)
users = session.query(User).options(joinedload(User.posts)).all()

# Решение 2: selectinload (отдельный SELECT для связанных)
users = session.query(User).options(selectinload(User.posts)).all()

# Решение 3: явный JOIN
from sqlalchemy.orm import contains_eager
users = (
    session.query(User)
    .join(User.posts)
    .options(contains_eager(User.posts))
    .all()
)

Миграции с Alembic

Для управления изменениями схемы БД я использовал Alembic:

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

# Файл: alembic/versions/001_add_phone_number.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column(
        'users',
        sa.Column('phone_number', sa.String(20), nullable=True)
    )
    op.create_index('ix_users_phone_number', 'users', ['phone_number'])

def downgrade():
    op.drop_index('ix_users_phone_number', table_name='users')
    op.drop_column('users', 'phone_number')

Работа с NoSQL: MongoDB

Для проектов, где требовалась гибкость схемы, я использовал MongoDB:

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['myapp']
collection = db['events']

# Вставка документов
event = {
    "user_id": 123,
    "event_type": "login",
    "timestamp": datetime.utcnow(),
    "metadata": {"ip": "192.168.1.1", "browser": "Chrome"}
}
collection.insert_one(event)

# Запросы с фильтрацией
events = collection.find({
    "event_type": "login",
    "timestamp": {"$gte": datetime(2024, 1, 1)}
})

# Агрегация
result = collection.aggregate([
    {"$match": {"event_type": "login"}},
    {"$group": {"_id": "$user_id", "count": {"$sum": 1}}},
    {"$sort": {"count": -1}}
])

Кеширование с Redis

Для повышения производительности я часто использовал Redis:

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Кеширование результатов
def get_user_with_cache(user_id):
    # Попыталась из кеша
    cached = redis_client.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    # Если нет — запрашиваем из БД
    user = session.query(User).filter(User.id == user_id).first()
    
    # Кешируем на 1 час
    redis_client.setex(
        f"user:{user_id}",
        3600,
        json.dumps(user.to_dict())
    )
    
    return user

# Использование Redis для очередей
redis_client.rpush('task_queue', json.dumps({'task': 'send_email', 'user_id': 123}))

Работа с миграциями и версионированием

Мои основные практики:

• Версионирование схемы: каждое изменение — отдельная миграция
• Обратимость: всегда писал downgrade() версии
• Тестирование миграций: проверял на тестовых данных
• Документирование: описывал причину каждого изменения

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

Я использовал несколько техник для повышения производительности:

# 1. Batch операции
users_data = [
    {"name": "John", "email": "john@example.com"},
    {"name": "Jane", "email": "jane@example.com"}
]
session.bulk_insert_mappings(User, users_data)

# 2. Индексы на часто используемых полях
from sqlalchemy import Index
Index('idx_user_email', User.email)

# 3. LIMIT и OFFSET для пагинации
page = 1
page_size = 20
users = session.query(User).limit(page_size).offset((page-1)*page_size).all()

# 4. Проверка плана выполнения
# EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'test@example.com'

Трансакции и блокировки

Для сложных операций я использовал транзакции:

from sqlalchemy import event
from sqlalchemy.orm import Session

try:
    # Начало транзакции
    user = session.query(User).filter(User.id == 1).first()
    user.balance -= 100
    
    another_user = session.query(User).filter(User.id == 2).first()
    another_user.balance += 100
    
    session.commit()  # Либо оба изменения, либо ничего
except Exception as e:
    session.rollback()  # Откат всех изменений
    raise

Заключение

Мой опыт с базами данных охватывает:

• Реляционные БД (PostgreSQL, MySQL)
• NoSQL (MongoDB)
• ORM (SQLAlchemy)
• Кеширование (Redis)
• Оптимизацию (индексы, explain plans)
• Миграции (Alembic)
• Масштабирование (шардирование, репликация)

Ключевой навык — выбор правильного инструмента для задачи и понимание performance implications каждого решения.