В каких случаях используют Hash в БД

# Использование Hash в базах данных

Определение

Hash в контексте БД — это структура данных, которая отображает ключи на значения. В SQL базах это часто означает хеш-таблицы или индексы, а в NoSQL (Redis, MongoDB) — это специальный тип данных для хранения key-value пар.

Основные случаи использования

1. Хеширование паролей (самый важный случай)

Пароли НИКОГДА не хранят в открытом виде. Вместо этого хранят хеш пароля.

import bcrypt
from sqlalchemy import Column, String, Integer, create_engine
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(100), unique=True)
    password_hash = Column(String(255))  # Хеш пароля, не сам пароль!
    email = Column(String(100), unique=True)
    
    def set_password(self, password: str):
        """Хешируем пароль перед сохранением"""
        # bcrypt добавляет соль и хеширует
        self.password_hash = bcrypt.hashpw(
            password.encode('utf-8'),
            bcrypt.gensalt(rounds=12)  # 12 раундов хеширования
        ).decode('utf-8')
    
    def check_password(self, password: str) -> bool:
        """Проверяем пароль против хеша"""
        return bcrypt.checkpw(
            password.encode('utf-8'),
            self.password_hash.encode('utf-8')
        )

# Использование
user = User(username='john')
user.set_password('my_secure_password_123')
session.add(user)
session.commit()

# При логине:
login_user = session.query(User).filter_by(username='john').first()
if login_user.check_password('my_secure_password_123'):
    print("Пароль верен!")
else:
    print("Неправильный пароль!")

Почему используется:

• Безопасность: даже если БД утечёт, пароли защищены
• Одностороннее: невозможно восстановить пароль из хеша
• Соль: каждый хеш уникален благодаря salt, поэтому два одинаковых пароля имеют разные хеши

2. Кэширование с помощью Redis Hash

Redis hash — это структура для хранения отображений, идеальна для кэширования.

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Сценарий: кэширование информации о пользователе
def get_user_info(user_id: int) -> dict:
    """Получает информацию о пользователе с кэшированием"""
    
    # Проверяем кэш (Redis hash)
    cached = redis_client.hgetall(f'user:{user_id}')
    
    if cached:
        print("Из кэша")
        return cached
    
    # Если кэша нет, получаем из БД
    user = db.query(User).get(user_id)
    
    if not user:
        return None
    
    user_data = {
        'id': str(user.id),
        'username': user.username,
        'email': user.email,
        'created_at': user.created_at.isoformat()
    }
    
    # Сохраняем в Redis hash на 1 час
    redis_client.hset(f'user:{user_id}', mapping=user_data)
    redis_client.expire(f'user:{user_id}', 3600)  # 1 час TTL
    
    return user_data

# Использование
user_data = get_user_info(123)
print(user_data)  # {'id': '123', 'username': 'john', ...}

# Redis hash выглядит так:
# HGETALL user:123
# => ["id", "123", "username", "john", "email", "john@example.com", ...]

3. Индексирование для быстрого поиска

БД используют хеш-индексы (hash indexes) для ускорения поиска по определённым полям.

from sqlalchemy import Index, Column, String, Integer

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    sku = Column(String(50), unique=True)  # Уникальный номер товара
    name = Column(String(200))
    price = Column(Float)
    
    # Создаём хеш-индекс для быстрого поиска по SKU
    __table_args__ = (
        Index('idx_product_sku', 'sku', mysql_using='HASH'),
    )

# Поиск по SKU очень быстрый благодаря хеш-индексу
product = session.query(Product).filter_by(sku='SKU-12345').first()
# Даже с миллионом товаров поиск будет O(1) в среднем

4. Дедупликация данных

Используют хеши для обнаружения дублирующихся данных.

import hashlib

class Document(Base):
    __tablename__ = 'documents'
    
    id = Column(Integer, primary_key=True)
    content = Column(Text)
    content_hash = Column(String(64), unique=True)  # SHA-256 хеш содержимого
    
    def calculate_hash(self):
        """Вычисляет SHA-256 хеш содержимого"""
        return hashlib.sha256(self.content.encode()).hexdigest()

# При добавлении документа
def add_document(content: str):
    # Вычисляем хеш
    doc_hash = hashlib.sha256(content.encode()).hexdigest()
    
    # Проверяем, есть ли уже документ с таким же содержимым
    existing = session.query(Document).filter_by(content_hash=doc_hash).first()
    
    if existing:
        print(f"Документ уже существует (ID: {existing.id})")
        return existing.id
    
    # Создаём новый документ
    doc = Document(content=content, content_hash=doc_hash)
    session.add(doc)
    session.commit()
    
    return doc.id

# Использование
id1 = add_document("Hello World")  # Новый документ
id2 = add_document("Hello World")  # Тот же хеш, вернёт id1

5. Session хранилище (в web приложениях)

Хеши используются для хранения session данных.

# Flask пример с Redis session store
import redis
from flask_session import Session
from flask import Flask, session

app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = redis.Redis(
    host='localhost',
    port=6379,
    decode_responses=True
)

Session(app)

@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    
    user = authenticate(username, password)
    
    if user:
        # Сохраняем session в Redis hash
        session['user_id'] = user.id
        session['username'] = user.username
        session['login_time'] = datetime.now().isoformat()
        
        # В Redis создаётся hash вроде:
        # HSET session:abc123def456 user_id 42 username john login_time 2024-01-15...
        
        return redirect('/dashboard')
    else:
        return "Login failed", 401

@app.route('/profile')
def profile():
    # Получаем session данные из Redis hash
    user_id = session.get('user_id')
    username = session.get('username')
    
    return f"Welcome {username}!"

6. API ключи и токены

Хеши для безопасного хранения API ключей.

class APIKey(Base):
    __tablename__ = 'api_keys'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    key_hash = Column(String(64), unique=True)  # Хеш ключа, не сам ключ
    prefix = Column(String(20))  # Для идентификации (показываем пользователю)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    @staticmethod
    def generate_key() -> str:
        """Генерирует случайный API ключ"""
        import secrets
        return secrets.token_urlsafe(32)
    
    @classmethod
    def create_key(cls, user_id: int):
        """Создаёт новый API ключ"""
        key = cls.generate_key()
        key_hash = hashlib.sha256(key.encode()).hexdigest()
        prefix = key[:8]  # Показываем первые 8 символов
        
        api_key = cls(
            user_id=user_id,
            key_hash=key_hash,
            prefix=prefix
        )
        
        session.add(api_key)
        session.commit()
        
        # Возвращаем только один раз (пользователь должен сохранить)
        return key
    
    @classmethod
    def verify_key(cls, key: str) -> Optional['APIKey']:
        """Проверяет API ключ"""
        key_hash = hashlib.sha256(key.encode()).hexdigest()
        return session.query(cls).filter_by(key_hash=key_hash).first()

# Использование
# При создании:
key = APIKey.create_key(user_id=42)
print(f"Ваш API ключ: {key}")  # Показываем один раз

# При использовании:
@app.route('/api/data')
def api_data():
    api_key = request.headers.get('Authorization', '').replace('Bearer ', '')
    
    key_record = APIKey.verify_key(api_key)
    
    if not key_record:
        return {"error": "Invalid API key"}, 401
    
    user = session.query(User).get(key_record.user_id)
    return {"data": "..."}

7. Bloom Filter для быстрой проверки наличия

Используют хеши для реализации Bloom Filter.

from pybloom import BloomFilter

# Пример: проверка, является ли email на чёрном списке
blacklist = BloomFilter(capacity=100000, error_rate=0.001)

# Добавляем "плохие" email'ы
blacklist.add('spam@example.com')
blacklist.add('bot@example.com')

# Быстрая проверка (O(k) где k - количество хешей)
if 'spam@example.com' in blacklist:
    print("Email на чёрном списке")
else:
    print("Email выглядит хорошо")

# Преимущества:
# - Очень мало памяти (100k элементов - 12KB)
# - O(1) поиск вместо O(log n) для B-tree
# - Минусы: возможны false positives, нельзя удалить элемент

8. Консистентное хеширование в распределённых системах

Для распределения данных по нескольким узлам.

from sortedcontainers import SortedDict
import hashlib

class ConsistentHash:
    def __init__(self, nodes=None, replicas=3):
        self.replicas = replicas  # Количество виртуальных узлов
        self.ring = SortedDict()
        self.nodes = set()
        
        if nodes:
            for node in nodes:
                self.add_node(node)
    
    def _hash(self, key: str) -> int:
        """Хешируем ключ в число"""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def add_node(self, node: str):
        """Добавляем узел в кольцо"""
        self.nodes.add(node)
        
        for i in range(self.replicas):
            virtual_key = f"{node}:{i}"
            hash_key = self._hash(virtual_key)
            self.ring[hash_key] = node
    
    def remove_node(self, node: str):
        """Удаляем узел из кольца"""
        self.nodes.discard(node)
        
        for i in range(self.replicas):
            virtual_key = f"{node}:{i}"
            hash_key = self._hash(virtual_key)
            del self.ring[hash_key]
    
    def get_node(self, key: str) -> str:
        """Получаем узел для ключа"""
        if not self.ring:
            return None
        
        hash_key = self._hash(key)
        
        # Ищем первый узел >= hash_key
        nodes = self.ring.bisect_right(hash_key)
        
        if nodes == len(self.ring):
            # Обходимся к началу кольца
            return self.ring[self.ring.keys()[0]]
        else:
            return self.ring[self.ring.keys()[nodes]]

# Использование
hash_ring = ConsistentHash(nodes=['server1', 'server2', 'server3'])

# При сохранении данных:
user_data = {'id': 123, 'name': 'John'}
server = hash_ring.get_node(f'user:123')
print(f"Сохраняем на {server}")  # => server2

# При добавлении нового сервера - переставляется минимум данных
hash_ring.add_node('server4')

Типы хеш-функций в БД

MD5 (не используется в production)

hashlib.md5(b'password').hexdigest()  # 5f4dcc3b5aa765d61d8327deb882cf99
# Ненадёжна, есть коллизии

SHA-256 (хорошо для большинства случаев)

hashlib.sha256(b'password').hexdigest()  # 5e884898da280047...
# Медленнее, но безопаснее

bcrypt (для паролей)

bcrypt.hashpw(b'password', bcrypt.gensalt())  # $2b$12$...
# Специально для паролей, адаптивна

Argon2 (лучший выбор для паролей)

from argon2 import PasswordHasher

ph = PasswordHasher()
hash = ph.hash("password")
# Современный, медленный, против GPU атак

Заключение

Хеши используются в БД для:

1. Безопасности паролей — bcrypt, Argon2
2. Кэширования — Redis hash структуры
3. Индексирования — hash индексы для O(1) поиска
4. Дедупликации — обнаружение дублей по хешу
5. Session хранилища — Redis hash для session данных
6. API ключей — безопасное хранение токенов
7. Bloom filters — быстрая проверка наличия
8. Распределённых систем — консистентное хеширование

Ключевое правило: никогда не хранить чувствительные данные (пароли, ключи) в открытом виде. Всегда хешировать с использованием современных алгоритмов.