Redis và FastAPI - khi nào nên lựa chọn?

I. Redis là gì?

Redis (Remote Dictionary Server) là một kho lưu trữ cấu trúc dữ liệu in-memory, hoạt động như một database, cache và message broker.

Đặc điểm quan trọng nhất của Redis:

• In-memory storage: Dữ liệu được lưu trong RAM, nên tốc độ cực nhanh, thường đạt 100,000+ tính toán/giây
• Persistence: Mặc dù lưu trong RAM, Redis vẫn có thể lưu dữ liệu xuống đĩa để đảm bảo dữ liệu không bị mất khi khởi động lại
• Đa dạng cấu trúc dữ liệu: Không chỉ có key-value đơn giản mà còn hỗ trợ Lists, Sets, Sorted Sets, Hashes, Streams,...

Các trường hợp sử dụng Redis

"Đừng dùng Redis như một cái búa và coi mọi vấn đề như cái đinh 😁"

• Caching: Lưu cache để tăng tốc ứng dụng
• Session storage: Lưu trữ phiên người dùng
• Real-time analytics: Phân tích dữ liệu thời gian thực
• Queues: Hàng đợi xử lý task
• Leaderboards/Counting: Bảng xếp hạng, đếm số lượng
• Pub/Sub messaging: Messaging giữa các components

Không dùng Redis khi:

1. Cần lưu trữ dữ liệu quan hệ phức tạp
2. Dữ liệu quá lớn so với bộ nhớ có sẵn
3. Cần ACID transactions đầy đủ
4. Không có kế hoạch quản lý bộ nhớ và persistence

Một số cấu trúc dữ liệu quan trọng trong Redis

II. Tích hợp Redis với FastAPI

Cài đặt Redis

# Cài đặt Redis (Linux/Mac)
# Ubuntu/Debian
sudo apt-get install redis-server

# macOS với Homebrew
brew install redis

# Cài đặt các thư viện Python
pip install fastapi uvicorn redis jinja2

Nếu sử dụng Windows, bạn có thể tải Redis từ https://github.com/tporadowski/redis/releases hoặc sử dụng Docker.

Mô hình kết nối cơ bản

Tạo file main.py

import redis
from fastapi import FastAPI

app = FastAPI()

# Kết nối tới Redis
r = redis.Redis(
    host='localhost',
    port=6379, # host và port: Địa chỉ của Redis server

    db=0,
    decode_responses=True  # tự động chuyển bytes sang string
)

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    # Đọc từ Redis
    item = r.get(f"item:{item_id}")
    if item is None:
        # Không tìm thấy trong Redis
        return {"error": "Item not found"}
    return {"item": item}

Chạy server: python main.py

Xử lý exception:

# Cách làm sai:
value = r.get("key")

# Cách làm đúng:
try:
    value = r.get("key")
except redis.RedisError as e:
    log.error(f"Redis error: {e}")
    # Xử lý fallback

Các pattern phổ biến khi dùng Redis với FastAPI

a. Caching API results

@app.get("/products/{product_id}")
async def get_product(product_id: int):
    # Thử lấy từ cache
    cache_key = f"product:{product_id}"
    cached_data = r.get(cache_key)
    
    if cached_data:
        return json.loads(cached_data)
    
    # Nếu không có trong cache, query database
    product = db.query(Product).filter(Product.id == product_id).first()
    
    if product:
        # Cache kết quả với TTL 1 giờ
        r.setex(cache_key, 3600, json.dumps(product.dict()))
        return product
    
    return {"error": "Product not found"}

b. Rate limiting

@app.get("/api/data")
async def get_data(request: Request):
    client_ip = request.client.host
    key = f"rate:limit:{client_ip}"
    
    # Increment counter
    current = r.incr(key)
    
    # Set expiry if this is the first request
    if current == 1:
        r.expire(key, 60)  # 60 seconds window
    
    # Check if rate limit exceeded
    if current > 10:  # Max 10 requests per minute
        return {"error": "Rate limit exceeded"}
    
    # Process request normally
    return {"data": "Here's your data!"}

c. Session storage

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    session_id = str(uuid.uuid4())
    
    # Lưu session vào Redis với TTL 1 giờ
    r.setex(f"session:{session_id}", 3600, json.dumps({"user_id": user.id}))
    
    return {"access_token": session_id}

async def get_current_user(token: str = Depends(oauth2_scheme)):
    session_data = r.get(f"session:{token}")
    if not session_data:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
        )
    
    user_data = json.loads(session_data)
    user = get_user(user_data["user_id"])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    return user

III. Lưu ý quan trọng khi dùng Redis

1. Bảo mật Redis

Redis mặc định không có authentication và có thể truy cập từ mọi nơi nếu không được cấu hình đúng:

Các biện pháp bảo mật:

• Đặt mật khẩu: requirepass trong redis.conf
• Bind địa chỉ IP cụ thể: bind 127.0.0.1
• Sử dụng Redis trong mạng riêng
• Sử dụng TLS cho kết nối từ xa
• Vô hiệu hóa các lệnh nguy hiểm: rename-command FLUSHALL ""

2. Quản lý bộ nhớ

Redis là in-memory nên việc quản lý bộ nhớ rất quan trọng:

"Đừng lưu những thứ không cần thiết vào Redis. RAM là tài nguyên quý giá."

• Cài đặt maxmemory và policy: volatile-lru, allkeys-lru
• Sử dụng TTL (EXPIRE) cho các key khi có thể
• Giám sát bộ nhớ sử dụng: INFO memory
• Sử dụng các lệnh tiết kiệm bộ nhớ như HSCAN, SSCAN thay vì HGETALL, SMEMBERS cho dữ liệu lớn

3. Hiệu suất

• Sử dụng pipeline để giảm độ trễ mạng:

pipe = r.pipeline()
pipe.set("key1", "value1")
pipe.set("key2", "value2")
pipe.set("key3", "value3")
pipe.execute()  # Gửi 3 lệnh trong 1 lần round-trip

• Sử dụng các lệnh atomic: INCR thay vì GET + SET
• Xem xét sử dụng Redis Cluster cho khả năng mở rộng

IV. Tổng kết

"Redis là công cụ mạnh mẽ, nhưng như mọi công cụ, việc biết khi nào sử dụng nó quan trọng hơn việc biết cách sử dụng nó." Để làm quen có thể bắt đầu với dự án đơn giản tạo ứng dụng học tiếng Anh: https://github.com/hasonsk/redis-fastapi-example

Tài liệu tham khảo

• https://redis.io/docs/latest