FastAPI狀態管理：簡單實現全局變量與緩存

from fastapi import FastAPI
import asyncio

app = FastAPI()

# 定義全局變量：用於存儲計數器
count = 0
# 定義異步鎖（解決多請求同時修改的競態條件）
lock = asyncio.Lock()

@app.get("/counter")
async def get_counter():
    global count
    async with lock:  # 獲取鎖，確保同一時間只有一個請求修改count
        count += 1
        return {"current_count": count}

from fastapi import FastAPI
import time

app = FastAPI()

# 模擬數據庫查詢（實際項目中可替換爲真實數據庫操作）
def get_user_from_db(user_id: int):
    time.sleep(1)  # 模擬1秒查詢延遲
    return {"user_id": user_id, "name": f"User_{user_id}"}

# 定義內存緩存字典
user_cache = {}

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    # 1. 先查緩存
    if user_id in user_cache:
        return {"source": "cache", "data": user_cache[user_id]}

    # 2. 緩存未命中，從數據庫查詢並緩存
    user_data = get_user_from_db(user_id)
    user_cache[user_id] = user_data  # 存入緩存
    return {"source": "database", "data": user_data}

from fastapi import FastAPI
from cachetools import LRUCache, TTLCache
from typing import Optional

app = FastAPI()

# 方式1：LRU緩存（最近最少使用，最多存儲10條數據）
user_cache_lru = LRUCache(maxsize=10)

# 方式2：TTL緩存（自動過期，10秒內未被訪問的數據自動清除）
user_cache_ttl = TTLCache(maxsize=10, ttl=10)

# 模擬用戶查詢（與前例相同）
def get_user_from_db(user_id: int):
    time.sleep(1)
    return {"user_id": user_id, "name": f"User_{user_id}"}

@app.get("/user/lru/{user_id}")
async def get_user_lru(user_id: int):
    # 檢查LRU緩存
    if user_id in user_cache_lru:
        return {"source": "lru_cache", "data": user_cache_lru[user_id]}
    user_data = get_user_from_db(user_id)
    user_cache_lru[user_id] = user_data
    return {"source": "database", "data": user_data}

@app.get("/user/ttl/{user_id}")
async def get_user_ttl(user_id: int):
    # 檢查TTL緩存（10秒過期）
    if user_id in user_cache_ttl:
        return {"source": "ttl_cache", "data": user_cache_ttl[user_id]}
    user_data = get_user_from_db(user_id)
    user_cache_ttl[user_id] = user_data
    return {"source": "database", "data": user_data}

from fastapi import FastAPI
import redis.asyncio as redis
import json

app = FastAPI()

# 連接Redis（需提前安裝redis：pip install redis）
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

@app.get("/user/redis/{user_id}")
async def get_user_redis(user_id: int):
    # 1. 嘗試從Redis緩存讀取
    cached_data = await redis_client.get(f"user:{user_id}")
    if cached_data:
        return {"source": "redis_cache", "data": json.loads(cached_data)}

    # 2. 緩存未命中，從數據庫查詢
    user_data = get_user_from_db(user_id)
    # 3. 存入Redis，設置10分鐘過期
    await redis_client.setex(
        f"user:{user_id}",  # 鍵：user:1
        600,                # 過期時間：600秒（10分鐘）
        json.dumps(user_data)  # 值：JSON序列化
    )
    return {"source": "database", "data": user_data}