Работа с базами данных в Python: полное руководство с примерами - Часть 3

# НЕПРАВИЛЬНО (уязвимо к SQL-инъекциям):
def unsafe_query(user_input):
    cursor.execute(f"SELECT * FROM users WHERE username = '{user_input}'")

# ПРАВИЛЬНО (использование параметризованных запросов):
def safe_query(user_input):
    cursor.execute("SELECT * FROM users WHERE username = ?", (user_input,))

import hashlib
import os
import binascii

def hash_password(password: str) -> str:
    """Хеширование пароля с использованием соли"""
    salt = hashlib.sha256(os.urandom(60)).hexdigest().encode('ascii')
    pwdhash = hashlib.pbkdf2_hmac(
        'sha512',
        password.encode('utf-8'),
        salt,
        100000
    )
    pwdhash = binascii.hexlify(pwdhash)
    return (salt + pwdhash).decode('ascii')

def verify_password(stored_password: str, provided_password: str) -> bool:
    """Проверка пароля"""
    salt = stored_password[:64]
    stored_password = stored_password[64:]
    pwdhash = hashlib.pbkdf2_hmac(
        'sha512',
        provided_password.encode('utf-8'),
        salt.encode('ascii'),
        100000
    )
    pwdhash = binascii.hexlify(pwdhash).decode('ascii')
    return pwdhash == stored_password

import logging
import time

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def query_logger(func):
    """Декоратор для логирования запросов"""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            execution_time = time.time() - start_time
            
            logger.info(
                f"Query executed: {func.__name__}, "
                f"Time: {execution_time:.3f}s"
            )
            
            # Логирование медленных запросов
            if execution_time > 1.0:
                logger.warning(
                    f"Slow query detected: {func.__name__}, "
                    f"Time: {execution_time:.3f}s"
                )
            
            return result
        except Exception as e:
            logger.error(f"Query failed: {func.__name__}, Error: {e}")
            raise
    return wrapper

# Использование
@query_logger
def get_user_tasks(user_id):
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM tasks WHERE user_id = ?", (user_id,))
        return cursor.fetchall()

import redis
import json
import pickle
from functools import wraps

class CacheManager:
    """Менеджер кэширования"""
    
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis = redis.Redis(host=host, port=port, db=db)
    
    def cache(self, timeout=300):
        """Декоратор для кэширования результатов функций"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Создание ключа кэша на основе аргументов функции
                cache_key = f"{func.__name__}:{args}:{kwargs}"
                
                # Попытка получить данные из кэша
                cached_data = self.redis.get(cache_key)
                if cached_data:
                    return pickle.loads(cached_data)
                
                # Выполнение функции, если данных нет в кэше
                result = func(*args, **kwargs)
                
                # Сохранение результата в кэш
                self.redis.setex(
                    cache_key,
                    timeout,
                    pickle.dumps(result)
                )
                
                return result
            return wrapper
        return decorator

# Использование
cache_manager = CacheManager()

@cache_manager.cache(timeout=600)  # Кэширование на 10 минут
def get_user_statistics(user_id):
    """Получение статистики пользователя (ресурсоемкая операция)"""
    # ... сложные вычисления или запросы к БД
    return statistics