Какие знаешь технологии валидации кеша?

Технологии валидации кеша

Валидация кеша — это механизм, который позволяет проверить, актуален ли закешированный контент без повторной загрузки всех данных. Это критично для HTTP кэширования и оптимизации сетевых запросов.

1. HTTP Headers — основной механизм

Cache-Control Header

from fastapi import FastAPI, Response
from datetime import datetime, timedelta

app = FastAPI()

@app.get("/api/data")
async def get_data(response: Response):
    data = {"message": "Hello"}
    
    # Кэш на 3600 секунд (1 час)
    response.headers["Cache-Control"] = "public, max-age=3600"
    
    # Или более сложное управление
    response.headers["Cache-Control"] = "public, max-age=3600, must-revalidate"
    
    # Параметры:
    # public — может кэшироваться везде (CDN, браузер)
    # private — только в браузере, не в CDN
    # no-cache — всегда проверяй валидность перед использованием
    # no-store — не кэшируй вообще
    # must-revalidate — после истечения обязательно проверить
    # immutable — контент никогда не изменяется
    
    return data

Директивы Cache-Control:

Cache-Control: public, max-age=31536000, immutable
  ↓
  Кэшируй везде, на 1 год, контент неизменяемый

Cache-Control: private, max-age=3600, must-revalidate
  ↓
  Только в браузере, 1 час, затем проверь валидность

Cache-Control: no-cache
  ↓
  Не используй кэш без проверки валидности

Cache-Control: no-store
  ↓
  Не кэшируй вообще (для чувствительных данных)

2. ETag — уникальный идентификатор контента

ETag (Entity Tag) — это хеш контента, который позволяет проверить, изменился ли он.

from fastapi import FastAPI, Response, Request
import hashlib

app = FastAPI()

DATA = {"message": "Hello World"}

def calculate_etag(data):
    """Рассчитай ETag как хеш содержимого"""
    import json
    content = json.dumps(data, sort_keys=True)
    return hashlib.md5(content.encode()).hexdigest()

@app.get("/api/data")
async def get_data(request: Request, response: Response):
    # Рассчитай ETag текущего контента
    etag = f'"{calculate_etag(DATA)}"'  # Кавычки обязательны!
    
    # Отправь ETag клиенту
    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "public, max-age=3600"
    
    # Если клиент отправил If-None-Match с тем же ETag
    if request.headers.get("If-None-Match") == etag:
        # Данные не изменились — верни 304 Not Modified
        response.status_code = 304
        return None  # Не отправляй тело
    
    return DATA

# Клиентская сторона (JavaScript)
"""
первый запрос:
GET /api/data
← 200 OK
← ETag: "abc123"
← [тело ответа]

сохраняем ETag

первый повторный запрос:
GET /api/data
If-None-Match: "abc123"
← 304 Not Modified
← (тело НЕ отправляется)

используем закешированные данные из браузера
"""

Weak ETag vs Strong ETag:

from fastapi import Response

@app.get("/resource")
async def get_resource(response: Response):
    # Strong ETag — для точного совпадения
    response.headers["ETag"] = '"123abc"'
    
    # Weak ETag — для приблизительного совпадения
    # (может отличаться в деталях, но суть одна)
    response.headers["ETag"] = 'W/"123abc"'
    
    # Weak используется для данных, которые немного могут отличаться
    # (например, время последнего доступа добавляется)
    
    return {"data": "content"}

3. Last-Modified — время последнего изменения

from fastapi import FastAPI, Response, Request
from datetime import datetime
import time

app = FastAPI()

resource_data = {"content": "Hello"}
last_modified_time = datetime.now()

@app.get("/api/resource")
async def get_resource(request: Request, response: Response):
    # Отправь время последнего изменения
    last_modified = last_modified_time.strftime("%a, %d %b %Y %H:%M:%S GMT")
    response.headers["Last-Modified"] = last_modified
    response.headers["Cache-Control"] = "public, max-age=3600"
    
    # Если клиент отправил If-Modified-Since
    if_modified_since = request.headers.get("If-Modified-Since")
    if if_modified_since == last_modified:
        # Контент не изменился с указанного времени
        response.status_code = 304
        return None
    
    return resource_data

# Клиентский код
"""
первый запрос:
GET /api/resource
← 200 OK
← Last-Modified: "Mon, 22 Mar 2025 10:00:00 GMT"
← [тело ответа]

повторный запрос:
GET /api/resource
If-Modified-Since: "Mon, 22 Mar 2025 10:00:00 GMT"
← 304 Not Modified
← (тело НЕ отправляется)
"""

4. Conditional Requests — условные запросы

from fastapi import FastAPI, Response, Request, status
import hashlib

app = FastAPI()

data = {"version": 1, "content": "Important Data"}

def get_etag():
    return f'"{hashlib.md5(str(data).encode()).hexdigest()}"'

@app.get("/data")
async def get_data(request: Request, response: Response):
    etag = get_etag()
    last_modified = "Mon, 22 Mar 2025 10:00:00 GMT"
    
    response.headers["ETag"] = etag
    response.headers["Last-Modified"] = last_modified
    response.headers["Cache-Control"] = "public, max-age=0, must-revalidate"
    
    # Проверка If-None-Match (ETag)
    if_none_match = request.headers.get("If-None-Match")
    if if_none_match and if_none_match == etag:
        response.status_code = status.HTTP_304_NOT_MODIFIED
        return
    
    # Проверка If-Modified-Since
    if_modified_since = request.headers.get("If-Modified-Since")
    if if_modified_since and if_modified_since == last_modified:
        response.status_code = status.HTTP_304_NOT_MODIFIED
        return
    
    return data

# PUT запрос с проверкой If-Match (оптимистичная блокировка)
@app.put("/data")
async def update_data(request: Request, response: Response, new_data: dict):
    current_etag = get_etag()
    
    # Клиент отправляет If-Match с ETag, который он знает
    if_match = request.headers.get("If-Match")
    
    if if_match is None:
        response.status_code = status.HTTP_412_PRECONDITION_FAILED
        return {"error": "If-Match header required"}
    
    if if_match != current_etag:
        # Данные изменились другим клиентом
        response.status_code = status.HTTP_412_PRECONDITION_FAILED
        return {"error": "Resource was modified, conflict"}
    
    # Обновления безопасно
    global data
    data = new_data
    
    response.headers["ETag"] = get_etag()
    return {"success": True}

5. Cache Busting — явная инвалидация кеша

# Подход 1: Версионирование контента
@app.get("/static/script-v1.2.3.js")
async def get_script(response: Response):
    # Версия в URL — кэшируется навечно
    response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
    return {"script": "alert('v1.2.3')"}

# Подход 2: Hash-based cache busting
import hashlib

def get_file_hash(filepath):
    with open(filepath, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()[:8]

# В шаблоне HTML:
# <script src="/static/app.{hash}.js"></script>
# <script src="/static/app.a1b2c3d4.js"></script>

# Подход 3: Query string versioning
@app.get("/api/data")
async def get_data(response: Response, v: int = 1):
    # GET /api/data?v=2 — будет запрошено заново
    response.headers["Cache-Control"] = "public, max-age=3600"
    return {"version": v, "data": "content"}

# Подход 4: Время последнего обновления
import os

@app.get("/assets/style.css")
async def get_css(response: Response):
    file_stat = os.stat('assets/style.css')
    modification_time = int(file_stat.st_mtime)
    
    # GET /assets/style.css?t=1711097400
    response.headers["Cache-Control"] = "public, max-age=3600"
    return {"css": "content"}

6. Redis/Memcached кэширование в приложении

from redis import Redis
import json
from datetime import timedelta

redis_client = Redis(host='localhost', port=6379, db=0)

class CacheManager:
    def __init__(self, redis_client, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl
    
    def get(self, key: str):
        """Получи данные из кэша"""
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    def set(self, key: str, value, ttl: int = None):
        """Сохрани данные в кэш"""
        ttl = ttl or self.ttl
        self.redis.setex(key, ttl, json.dumps(value))
    
    def invalidate(self, key: str):
        """Инвалидируй кэш"""
        self.redis.delete(key)
    
    def invalidate_pattern(self, pattern: str):
        """Инвалидируй кэш по паттерну"""
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys)

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

@app.get("/api/user/{user_id}")
async def get_user(user_id: int):
    cache_key = f"user:{user_id}"
    
    # Попробуй кэш
    cached_user = cache.get(cache_key)
    if cached_user:
        return cached_user
    
    # Получи из БД
    user = await db.fetch_user(user_id)
    
    # Сохрани в кэш на 1 час
    cache.set(cache_key, user, ttl=3600)
    
    return user

@app.put("/api/user/{user_id}")
async def update_user(user_id: int, data: dict):
    # Обнови в БД
    updated_user = await db.update_user(user_id, data)
    
    # Инвалидируй кэш
    cache.invalidate(f"user:{user_id}")
    cache.invalidate(f"users:list:*")  # Инвалидируй список пользователей
    
    return updated_user

7. Cache Layers — многоуровневое кэширование

from functools import lru_cache
import asyncio

class MultiLayerCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.memory_cache = {}  # L1: в памяти приложения
    
    async def get(self, key: str):
        # L1: память приложения (очень быстро)
        if key in self.memory_cache:
            return self.memory_cache[key]
        
        # L2: Redis (быстро, но медленнее чем память)
        cached = self.redis.get(key)
        if cached:
            value = json.loads(cached)
            # Загрузи в L1
            self.memory_cache[key] = value
            return value
        
        # L3: БД (медленно)
        return None
    
    async def set(self, key: str, value, ttl: int = 3600):
        # Сохрани во все слои
        self.memory_cache[key] = value  # L1
        self.redis.setex(key, ttl, json.dumps(value))  # L2
    
    async def invalidate(self, key: str):
        # Инвалидируй все слои
        if key in self.memory_cache:
            del self.memory_cache[key]
        self.redis.delete(key)

# Декоратор для автоматического кэширования
def cached(ttl: int = 3600):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            # Составь ключ кэша
            cache_key = f"{func.__name__}:{args}:{kwargs}"
            
            # Попробуй кэш
            cached_result = await cache_manager.get(cache_key)
            if cached_result is not None:
                return cached_result
            
            # Вычисли результат
            result = await func(*args, **kwargs)
            
            # Сохрани в кэш
            await cache_manager.set(cache_key, result, ttl)
            
            return result
        return wrapper
    return decorator

@cached(ttl=3600)
async def expensive_operation(user_id: int):
    # Долгая операция
    return await complex_query(user_id)

Таблица сравнения технологий валидации

Best Practices

1. Статические ресурсы — используй Cache-Control: immutable, max-age=31536000
2. Динамический контент — используй ETag + Cache-Control: must-revalidate
3. Приватные данные — Cache-Control: private, no-store
4. Многоуровневое кэширование — память → Redis → БД
5. Явная инвалидация — при обновлении данных инвалидируй кэш
6. Версионируй API — помогает управлять кэшем между релизами
7. Мониторь хит-рейт кэша — определяй эффективность