Как реализовать кэширование?

Как реализовать кэширование

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

1. Python встроенное кэширование

@lru_cache (Least Recently Used):

from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_computation(n):
    print(f"Computing {n}")
    return n ** 2

print(expensive_computation(5))  # Computing 5 → 25
print(expensive_computation(5))  # Из кэша → 25
print(expensive_computation(10)) # Computing 10 → 100

@cache (без размера):

from functools import cache

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

2. Redis — рекомендуемый кэш

Установка:

pip install redis

Базовое использование:

import redis
import json

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

# Сохранение
redis_client.set('user:123', json.dumps({'name': 'John', 'age': 30}), ex=3600)  # 1 час

# Получение
value = redis_client.get('user:123')
if value:
    user = json.loads(value)
    print(user)  # {'name': 'John', 'age': 30}

Паттерн cache-aside:

def get_user(user_id):
    # 1. Проверить кэш
    cached = redis_client.get(f'user:{user_id}')
    if cached:
        return json.loads(cached)
    
    # 2. Если нет в кэше, получить из БД
    user = db.query(User).filter(User.id == user_id).first()
    
    # 3. Сохранить в кэш
    redis_client.set(f'user:{user_id}', json.dumps(user.dict()), ex=3600)
    
    return user

3. Django кэширование

Настройка Redis:

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
        }
    }
}

Использование в Django:

from django.core.cache import cache
from django.views.decorators.cache import cache_page

# Декоратор для вьюх (кэш на 5 минут)
@cache_page(60 * 5)
def user_list(request):
    users = User.objects.all()
    return render(request, 'users.html', {'users': users})

# Ручное кэширование
def get_user(user_id):
    cache_key = f'user:{user_id}'
    user = cache.get(cache_key)
    
    if user is None:
        user = User.objects.get(id=user_id)
        cache.set(cache_key, user, timeout=3600)  # 1 час
    
    return user

4. FastAPI кэширование

С Redis:

from fastapi import FastAPI
from fastapi_cache2 import FastAPICache2
from fastapi_cache2.backends.redis import RedisBackend
from fastapi_cache2.decorators import cache
from redis import asyncio as aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost")
    FastAPICache2.init(RedisBackend(redis), prefix="fastapi-cache")

@app.get("/users/{user_id}")
@cache(expire=3600)
async def get_user(user_id: int):
    # Кэшируется автоматически
    user = await db.get_user(user_id)
    return user

5. Memcached

Установка:

pip install pymemcache

Использование:

from pymemcache.client.base import Client

client = Client(('localhost', 11211))

# Установка
client.set('key', b'value', expire=3600)

# Получение
value = client.get('key')
if value:
    print(value.decode())  # value

6. In-Memory кэш

Простой вариант:

from datetime import datetime, timedelta

class SimpleCache:
    def __init__(self):
        self.store = {}
    
    def get(self, key):
        if key in self.store:
            value, expires_at = self.store[key]
            if expires_at > datetime.now():
                return value
            else:
                del self.store[key]  # Удалить истёкший
        return None
    
    def set(self, key, value, expire_seconds=3600):
        expires_at = datetime.now() + timedelta(seconds=expire_seconds)
        self.store[key] = (value, expires_at)

cache = SimpleCache()
cache.set('user:123', {'name': 'John'}, expire_seconds=3600)
print(cache.get('user:123'))  # {'name': 'John'}

7. Декоратор кэширования

from functools import wraps
from datetime import datetime, timedelta

def cached(expire_seconds=3600):
    def decorator(func):
        cache = {}
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(kwargs.items()))
            
            if key in cache:
                value, expires_at = cache[key]
                if expires_at > datetime.now():
                    return value
                else:
                    del cache[key]
            
            result = func(*args, **kwargs)
            cache[key] = (result, datetime.now() + timedelta(seconds=expire_seconds))
            return result
        
        return wrapper
    return decorator

@cached(expire_seconds=300)
def expensive_function(x):
    print(f"Computing {x}")
    return x ** 2

print(expensive_function(5))  # Computing 5 → 25
print(expensive_function(5))  # Из кэша → 25

8. Query кэширование

from django.db.models import Q

# Кэширование результата запроса
def get_active_users():
    cache_key = 'active_users'
    users = cache.get(cache_key)
    
    if users is None:
        users = User.objects.filter(is_active=True)
        cache.set(cache_key, list(users), timeout=3600)
    
    return users

# Инвалидация кэша при обновлении
def update_user(user_id, name):
    user = User.objects.get(id=user_id)
    user.name = name
    user.save()
    
    # Инвалидировать кэш
    cache.delete('active_users')
    cache.delete(f'user:{user_id}')

9. HTTP кэширование

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()

@app.get("/data")
async def get_data(response: Response):
    # Кэшировать на клиенте
    response.headers["Cache-Control"] = "public, max-age=3600"
    response.headers["ETag"] = '"123456"'
    
    return {"data": "value"}

10. Cache Layers

# Многоуровневое кэширование
def get_user_optimized(user_id):
    # L1: локальный кэш (быстро)
    user = local_cache.get(f'user:{user_id}')
    if user:
        return user
    
    # L2: Redis (среднее)
    user = redis.get(f'user:{user_id}')
    if user:
        local_cache.set(f'user:{user_id}', user, expire=60)
        return user
    
    # L3: Database (медленно)
    user = db.get_user(user_id)
    redis.set(f'user:{user_id}', user, expire=3600)
    local_cache.set(f'user:{user_id}', user, expire=60)
    
    return user

Стратегии инвалидации

# 1. Time-based (TTL)
cache.set('key', value, expire=3600)  # Автоматически удалится через час

# 2. Event-based
def on_user_updated(user_id):
    cache.delete(f'user:{user_id}')
    cache.delete('all_users')

# 3. LRU
@lru_cache(maxsize=128)  # Удаляет самые старые при превышении
def func():
    pass

# 4. Manual
cache.delete('specific_key')
cache.clear()  # Очистить все

Вывод: Кэширование критично для production приложений. Выбор инструмента зависит от needs:

• Функции: @lru_cache
• Веб-приложение: Redis
• Простой прототип: In-memory кэш
• Распределённая система: Redis/Memcached