Настройка кэша для запросов к базе данных через Redis

Представьте себе гиперактивный API, который шлёт запросы на сервер каждые полсекунды. Если каждый такой запрос должен обращаться к базе данных для получения данных, сервер рано или поздно начнёт паниковать, а ваш DevOps начнёт пить слишком много кофе. Вот тут и приходит на помощь кэширование! Вместо того чтобы каждый раз лезть в базу, сервис может временно сохранять результаты частых запросов в Redis.

Кэширование подходит для данных, которые:

• часто запрашиваются (например, популярные продукты в интернет-магазине),
• редко изменяются (например, курсы валют, которые обновляются раз в сутки),
• занимают немного места (Redis хранит данные в оперативной памяти, и её объём ограничен).

Однако избегайте кэширования динамически изменяющихся данных, так как это может привести к устаревшим или некорректным данным.

Определение TTL

TTL (Time to Live) — это время, в течение которого данные остаются актуальными в кэше. После его истечения данные удаляются. Для каждого типа данных нужно подбирать свой TTL:

• Быстро обновляемые данные: 30-60 секунд.
• Редко изменяемые данные: 5-10 минут.
• Почти статические данные (например, географическая информация): часы или даже дни.

---

Настройка кэширования для запроса

Пусть у нас есть база данных PostgreSQL, в которой хранится список пользователей. Задача: создать эндпоинт, который возвращает всех пользователей, и кэшировать этот запрос в Redis.

Сначала реализуем простой эндпоинт без кэширования:

# app.py
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from models import User  # Это модель SQLAlchemy

# Настройка FastAPI и базы данных
app = FastAPI()
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydatabase"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@app.get("/users")
async def get_users():
    async with async_session() as session:
        # Запрос всех пользователей из базы данных
        result = await session.execute("SELECT * FROM users")
        users = result.fetchall()
        return {"users": [dict(row) for row in users]}

Этот эндпоинт работает, но каждый раз обращается к базе данных.

Теперь добавим Redis для кэширования. Для этого установим библиотеку redis-py:

pip install redis

Настроим подключение к Redis и реализуем кэширование:

import redis
import json

# Подключение к Redis
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

@app.get("/users")
async def get_users():
    # Попробуем найти данные в кэше
    cached_users = redis_client.get("users")
    if cached_users:
        # Если данные найдены, вернём их
        return {"users": json.loads(cached_users)}

    # Если данных в кэше нет, делаем запрос к базе данных
    async with async_session() as session:
        result = await session.execute("SELECT * FROM users")
        users = result.fetchall()
        # Сохраняем результат в кэше на 60 секунд
        redis_client.setex("users", 60, json.dumps([dict(row) for row in users]))
        return {"users": [dict(row) for row in users]}

В этом коде:

1. Мы сначала проверяем, есть ли ключ "users" в Redis.
2. Если ключ найден, берём данные из кэша.
3. Если нет, запрашиваем данные из базы данных и сохраняем их в Redis.

Теперь запрос к /users будет работать гораздо быстрее!

---

Оптимизация кэширования данных

Для кэширования более сложных запросов (например, с параметрами) нужно динамически генерировать уникальный ключ.

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    key = f"user:{user_id}"  # Генерируем уникальный ключ
    cached_user = redis_client.get(key)
    if cached_user:
        return {"user": json.loads(cached_user)}

    async with async_session() as session:
        result = await session.execute(f"SELECT * FROM users WHERE id = {user_id}")
        user = result.fetchone()
        if user:
            redis_client.setex(key, 60, json.dumps(dict(user)))
        return {"user": dict(user) if user else None}

Здесь ключ кэша зависит от user_id. Это гарантирует, что кэш сохраняет данные для конкретного пользователя.

---

Инвалидирование кэша

Когда данные в базе данных обновляются, соответствующие данные в кэше становятся устаревшими. Чтобы избежать этого, необходимо сбрасывать (инвалидировать) кэш.

Представьте, что у нас есть эндпоинт для обновления пользователя. После обновления необходимо удалить старый кэш:

@app.put("/users/{user_id}")
async def update_user(user_id: int, user_data: dict):
    async with async_session() as session:
        # Пример обновления данных в базе
        await session.execute(
            f"UPDATE users SET name = :name WHERE id = {user_id}",
            {"name": user_data["name"]},
        )
        await session.commit()

    # Удаляем устаревший кэш
    redis_client.delete(f"user:{user_id}")
    return {"message": "User updated successfully"}

Теперь, после обновления пользователя, данные в кэше удаляются, и при следующем запросе будут обновлены.

---

Пример завершённого кэширования для списка

Работа с Redis становится ещё более гибкой с использованием функций для общего управления кэшем:

def get_or_set_cache(key, fetch_func, ttl=60):
    # Попробуем найти ключ в кэше
    cached_data = redis_client.get(key)
    if cached_data:
        return json.loads(cached_data)
    
    # Если такого ключа нет, получаем данные и кэшируем
    data = fetch_func()
    redis_client.setex(key, ttl, json.dumps(data))
    return data

@app.get("/users")
async def get_users():
    async def fetch_users():
        async with async_session() as session:
            result = await session.execute("SELECT * FROM users")
            return [dict(row) for row in result.fetchall()]

    # Используем обёртку для кэширования
    users = get_or_set_cache("users", fetch_users)
    return {"users": users}

Эта реализация позволяет переиспользовать логику кэширования в разных частях приложения.

---

Подводим итоги

Использование Redis для кэширования запросов к базе данных снижает нагрузку на сервер и ускоряет выполнение запросов. Мы научились:

• Подключать Redis к FastAPI.
• Настраивать кэш для запросов с использованием setex и TTL.
• Реализовывать динамическое кэширование с использованием уникальных ключей.
• Обновлять кэш при изменении данных в базе.

Поздравляю, вы только что сделали ваше приложение значительно быстрее! Однако, чтобы не перегружать ваш Redis, не забудьте про очистку старых и ненужных данных. Впереди нас ждёт ещё больше оптимизаций и секретов кэширования!