Оптимізація продуктивності FastAPI із використанням Redis

Як кешування допомагає пришвидшити роботу FastAPI? Уявіть, що ваш FastAPI-додаток — це ресторан. Щоразу, коли клієнт замовляє страву, кухня (ваша база даних) має приготувати її з нуля. Це потребує часу й ресурсів. Redis, у свою чергу, виступає в ролі супер-швидких офіціантів, які відразу подають страву клієнтам, якщо вона вже готова й гаряча.

Переваги Redis для оптимізації FastAPI:

1. Миттєвий доступ до даних: Redis зберігає дані в оперативній пам'яті, що робить його значно швидшим за традиційні бази даних.
2. Зниження навантаження на основну базу даних: кешування частих запитів звільняє вашу базу від виконання однакових операцій.
3. Підтримка TTL: механізм Time-To-Live дозволяє кешу автоматично очищуватись від застарілих даних.
4. Гнучкість у налаштуванні політик кешування: наприклад, можна застосувати стратегію «витіснення» рідко використовуваних даних.

---

Інтеграція Redis для повної оптимізації

Давайте розберемо основні компоненти оптимізації FastAPI з Redis, поєднаємо всі раніше вивчені техніки і застосуємо їх у реальному проєкті.

1. Налаштування оточення

Спочатку переконайтесь, що у вас встановлений Redis і підключена бібліотека aioredis у ваш FastAPI проєкт:

pip install aioredis

Якщо Redis ще не запущений, запустіть його через Docker:

docker run -d --name redis-server -p 6379:6379 redis

2. Налаштування підключення

Створюємо підключення до Redis у нашому FastAPI проєкті в окремому модулі redis_config.py:

import aioredis

# Пул підключень Redis
async def get_redis_pool():
    return await aioredis.from_url(
        "redis://localhost",
        encoding="utf-8",
        decode_responses=True
    )

Тепер налаштовуємо життєвий цикл додатка з Redis: підключення при старті і відключення при завершенні роботи:

from fastapi import FastAPI
from redis_config import get_redis_pool

app = FastAPI()
redis_client = None

@app.on_event("startup")
async def startup():
    global redis_client
    redis_client = await get_redis_pool()

@app.on_event("shutdown")
async def shutdown():
    if redis_client:
        await redis_client.close()

3. Кешування складного обчислювального запиту

Припустимо, у нашому застосунку є запит, який виконує важку аналітичну операцію над даними. Реалізуємо кешування для нього:

Модель:

from pydantic import BaseModel

class Product(BaseModel):
    id: int
    name: str
    price: float

Логіка маршруту:

from fastapi import Depends
import json  # Для серіалізації об'єктів

@app.get("/products/{product_id}")
async def get_product(product_id: int, redis=Depends(lambda: redis_client)):
    # Перевіряємо кеш
    cached_product = await redis.get(f"product:{product_id}")
    if cached_product:
        # Якщо дані вже є в кеші, повертаємо їх
        return json.loads(cached_product)

    # Якщо немає даних у кеші, виконуємо «важкі обчислення»
    product = {
        "id": product_id,
        "name": f"Product {product_id}",
        "price": 99.99,
    }

    # Зберігаємо результат у кеш на 60 секунд
    await redis.setex(f"product:{product_id}", 60, json.dumps(product))
    return product

Що тут відбувається?

1. Під час отримання запиту до API спочатку перевіряється, чи існує об'єкт у кеші.
2. Якщо об'єкт знайдено, він повертається миттєво, без звернення до бази даних.
3. Якщо ні, дані опрацьовуються (або запитуються з БД), а результат записується в Redis з TTL у 60 секунд.

---

Кешування списку (або сторінки даних)

Завдання: кешувати результати пагінації даних, щоб не генерувати їх щоразу. Розглянемо приклад з постами блогу:

@app.get("/posts")
async def get_posts(page: int = 1, redis=Depends(lambda: redis_client)):
    cache_key = f"posts:page:{page}"

    # Перевіряємо кеш
    cached_posts = await redis.get(cache_key)
    if cached_posts:
        return json.loads(cached_posts)

    # Якщо даних у кеші немає, імітуємо «важку» операцію
    posts = [{"id": i, "title": f"Post {i}"} for i in range((page-1)*10, page*10)]

    # Зберігаємо результат у кеш на 120 секунд
    await redis.setex(cache_key, 120, json.dumps(posts))
    return posts

---

Кешування залежності на рівні маршруту

FastAPI підтримує кешування через залежності. Наприклад, ми можемо винести кешування в окрему функцію:

from fastapi import Depends, HTTPException
from typing import Callable

async def cache_dependency(key: str, operation: Callable, ttl: int = 60):
    cached_value = await redis_client.get(key)
    if cached_value:
        return json.loads(cached_value)
    
    result = await operation()
    await redis_client.setex(key, ttl, json.dumps(result))
    return result

@app.get("/expensive-data")
async def expensive_operation(data=Depends(lambda: cache_dependency(
    key="expensive_data_cache",
    operation=lambda: mock_database_fetch(),
    ttl=300
))):
    return data

async def mock_database_fetch():
    # Емуляція запиту до бази даних
    return {"data": "This is heavy data processing"}

---

Моніторинг і налаштування Redis

Redis надає команди для моніторингу продуктивності:

• INFO — отримання загальної інформації про сервер.
• MONITOR — відображення всіх виконуваних команд у реальному часі.
• MEMORY USAGE key — перевірка використання пам'яті для конкретного ключа.

Підтримуйте оптимальні налаштування Redis, слідкуючи за обмеженнями оперативної пам'яті та обираючи підхожу політику витіснення даних, наприклад, LRU (Least Recently Used).

---

Візуалізація покращень продуктивності

Уявімо простий сценарій із вимірюванням часу відповіді FastAPI:

import time

start_time = time.time()
response = await client.get("/products/1")
cached_time = time.time() - start_time
print(f"Response Time (with Redis cache): {cached_time:.5f} seconds")

Експеримент з і без кешування покаже драматичне зменшення часу обробки при використанні Redis. Для навантажувального тестування можна використовувати інструменти, такі як Apache Benchmark (ab) або wrk.

---

Реалізація автоматичного інвалідування кешу

Коли дані оновлюються, потрібно видаляти застарілі елементи з кешу. Припустимо, у нашому застосунку оновлюється інформація про продукт:

@app.put("/products/{product_id}")
async def update_product(product_id: int, product: Product, redis=Depends(lambda: redis_client)):
    # Логіка оновлення даних (наприклад, у БД)

    # Видаляємо застарілий кеш
    await redis.delete(f"product:{product_id}")

    return {"message": "Product updated successfully"}

---

Підсумкова практика

Ваше завдання: інтегруйте Redis у свій FastAPI застосунок, протестуйте його під реальною навантаженням, а потім виміряйте час відповіді до і після впровадження кешування. Обов'язково поділіться результатами в Slack групі — давайте подивимось, чий застосунок стає ракетою!

Документація Redis

Документація aioredis