В какой момент нужно кэшировать данные

В какой момент кэшировать данные: стратегия и практика

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

Основные критерии для кэширования

1. Высокая стоимость получения данных

• Дорогие операции БД (JOIN, агрегация по большим таблицам)
• Внешние API запросы (сетевая задержка)
• Сложные вычисления (обработка больших объёмов данных)
• Парсинг/трансформация данных

2. Высокая частота обращений к одним и тем же данным

• Одни данные запрашиваются множеством пользователей (публичная информация, справочники)
• Повторяющиеся запросы в одну сессию
• Данные, которые редко меняются

3. Допустимость небольшой стараелости (stale data)

• Справочники, категории, налоги — обновляются редко
• User profile в рамках сессии
• Статистика, аналитика — допустима задержка в минуты/часы

Когда НЕ кэшировать

Критические случаи:

• Финансовые операции (баланс, платежи, транзакции)
• Состояние аутентификации и разрешений (auth, permissions)
• Персональные данные, которые часто меняются
• Данные с требованиями ACID

Уровни кэширования и их применение

1. Кэш на уровне приложения (in-memory)

from functools import lru_cache
import time

# ✅ Простое решение для справочников
@lru_cache(maxsize=128)
def get_tax_rate(country_code: str) -> float:
    # дорогой запрос в БД
    return query_tax_rate(country_code)

# ✅ С явным контролем TTL
class CachedRepository:
    def __init__(self):
        self._cache = {}
        self._timestamps = {}
    
    def get(self, key: str, ttl_seconds: int = 300) -> Optional[Any]:
        if key in self._cache:
            age = time.time() - self._timestamps[key]
            if age < ttl_seconds:
                return self._cache[key]
            del self._cache[key]
        
        value = self._fetch_from_source(key)
        self._cache[key] = value
        self._timestamps[key] = time.time()
        return value

2. Кэш на уровне БД (Redis, Memcached)

# ✅ Использование Redis для распределённого кэша
import redis
from json import dumps, loads

class RedisCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def get_user_profile(self, user_id: int, ttl: int = 3600):
        cache_key = f"user:{user_id}:profile"
        
        # Попытка получить из кэша
        cached = self.redis.get(cache_key)
        if cached:
            return loads(cached)
        
        # Если нет — запрос в БД
        user = db.query(User).get(user_id)
        
        # Сохраняем в кэш
        self.redis.setex(cache_key, ttl, dumps(user.to_dict()))
        
        return user.to_dict()

3. Кэш на уровне HTTP

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()

# ✅ Кэширование публичных данных
@app.get("/api/categories")
async def get_categories(response: Response):
    response.headers["Cache-Control"] = "public, max-age=3600"
    return await db.fetch_categories()

Практический пример: слои кэширования

class DataService:
    def __init__(self, db, redis_client):
        self.db = db
        self.redis = redis_client
        self._memory_cache = {}  # L1: In-memory
    
    async def get_product(self, product_id: int) -> Product:
        # L1: Memory cache (очень быстро, но локально)
        if product_id in self._memory_cache:
            return self._memory_cache[product_id]
        
        # L2: Redis (быстро, распределённо)
        cache_key = f"product:{product_id}"
        cached = await self.redis.get(cache_key)
        if cached:
            product = Product.from_dict(loads(cached))
            self._memory_cache[product_id] = product  # кэшируем в памяти
            return product
        
        # L3: БД (медленно)
        product = await self.db.get_product(product_id)
        
        # Заполняем обратно все слои
        await self.redis.setex(cache_key, 3600, dumps(product.to_dict()))
        self._memory_cache[product_id] = product
        
        return product

Инвалидация кэша

Когда инвалидировать:

• При создании/обновлении/удалении данных
• По расписанию (TTL)
• На основе событий

# ✅ Инвалидация при обновлении
async def update_product(self, product_id: int, data: dict):
    # Обновляем в БД
    product = await self.db.update_product(product_id, data)
    
    # Инвалидируем кэш
    await self.redis.delete(f"product:{product_id}")
    if product_id in self._memory_cache:
        del self._memory_cache[product_id]
    
    return product

# ✅ Soft invalidation (запрос на свежесть)
async def get_with_validation(self, product_id: int) -> Product:
    cached = await self.redis.get(f"product:{product_id}")
    if cached:
        # Проверяем в фоне, нужны ли обновления
        asyncio.create_task(self._validate_product(product_id))
        return Product.from_dict(loads(cached))
    return await self.get_product(product_id)

Резюме

Кэшируй когда:

• Операция дорогая (БД, внешние API, вычисления)
• Частые повторяющиеся обращения
• Допустима небольшая стараелость данных

Не кэшируй:

• Критические данные, требующие актуальности
• Редко запрашиваемые данные (потеря памяти)
• Данные с высокими требованиями к консистентности

Ключ к успеху — правильный выбор уровня кэширования, контроль TTL и стратегия инвалидации.