Как реализовать кэширование с Redis?python-46

pip install redis

import redis

# Создание подключения
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    password=None,  # если требуется
    socket_timeout=5
)

# Проверка подключения
try:
    redis_client.ping()
    print("Успешное подключение к Redis")
except redis.ConnectionError:
    print("Ошибка подключения к Redis")

# Запись в кэш
redis_client.set('user:1001', '{"name": "John", "age": 30}', ex=3600)  # ex - время жизни в секундах

# Чтение из кэша
user_data = redis_client.get('user:1001')
if user_data:
    print(user_data.decode('utf-8'))

from functools import wraps
import pickle
import hashlib

def cache_with_redis(redis_client, timeout=3600):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Создаем уникальный ключ на основе аргументов функции
            key = f"func:{func.__name__}:{hashlib.md5(pickle.dumps((args, kwargs))).hexdigest()}"

            # Пытаемся получить результат из кэша
            cached_result = redis_client.get(key)
            if cached_result is not None:
                return pickle.loads(cached_result)

            # Если в кэше нет, выполняем функцию
            result = func(*args, **kwargs)

            # Сохраняем результат в кэш
            redis_client.setex(key, timeout, pickle.dumps(result))

            return result
        return wrapper
    return decorator

# Пример использования
@cache_with_redis(redis_client, timeout=600)
def expensive_operation(user_id):
    # Тяжелые вычисления или запрос к БД
    return f"Result for user {user_id}"

# Сохраняем данные пользователя как хэш
redis_client.hset(
    'user:1001',
    mapping={
        'name': 'John',
        'age': '30',
        'email': 'john@example.com'
    }
)

# Устанавливаем TTL для всего хэша
redis_client.expire('user:1001', 3600)

# Получаем данные
user_name = redis_client.hget('user:1001', 'name')
print(user_name.decode('utf-8'))

def get_user_data(user_id):
    cache_key = f'user:{user_id}'
    cached_data = redis_client.get(cache_key)

    if cached_data:
        return cached_data.decode('utf-8')

    # Если нет в кэше, получаем из БД
    db_data = get_from_database(user_id)  # Предполагаемая функция

    # Сохраняем в кэш
    redis_client.setex(cache_key, 3600, db_data)

    return db_data

def update_user_data(user_id, new_data):
    # Обновляем данные в БД
    update_in_database(user_id, new_data)  # Предполагаемая функция

    # Инвалидируем кэш
    redis_client.delete(f'user:{user_id}')

def get_product_details(product_id):
    cache_key = f'product:{product_id}'

    # Пытаемся получить из кэша
    product_data = redis_client.get(cache_key)
    if product_data:
        return json.loads(product_data)

    # Если нет в кэше, загружаем из БД
    db_data = query_database_for_product(product_id)

    if db_data:
        # Сохраняем в кэш на 1 час
        redis_client.setex(cache_key, 3600, json.dumps(db_data))

    return db_data

# Создаем pipeline
pipe = redis_client.pipeline()

# Добавляем команды
for user_id in range(1000, 1010):
    pipe.set(f'user:{user_id}', f'user_data_{user_id}', ex=3600)

# Выполняем все команды за один round-trip
pipe.execute()