#FastAPI与Redis集成完全指南

# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    redis_url: str = "redis://localhost:6379/0"
    redis_max_connections: int = 20
    redis_socket_keepalive: bool = True
    
    class Config:
        env_file = ".env"

@lru_cache()
def get_settings():
    return Settings()

# redis_client.py
import redis.asyncio as redis
from config import get_settings

settings = get_settings()

class RedisManager:
    _client: redis.Redis | None = None

    @classmethod
    async def init(cls):
        """初始化连接池"""
        if not cls._client:
            cls._client = redis.from_url(
                settings.redis_url,
                max_connections=settings.redis_max_connections,
                socket_keepalive=settings.redis_socket_keepalive,
                decode_responses=True,  # 自动解码bytes为str
                encoding="utf-8"
            )
            await cls._client.ping()
            print("✅ Redis连接成功")
    
    @classmethod
    def get_client(cls) -> redis.Redis:
        if not cls._client:
            raise RuntimeError("Redis未初始化")
        return cls._client

# main.py生命周期绑定
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    await RedisManager.init()
    yield
    await RedisManager.get_client().close()

app = FastAPI(lifespan=lifespan)

# cache/decorators.py
import json
import hashlib
from functools import wraps
from redis.asyncio import Redis
from redis_client import RedisManager

def cached(
    key_prefix: str,
    ttl: int = 300,  # 默认5分钟
):
    """
    支持参数自动哈希的缓存装饰器
    用法：
    @cached("user:profile:{user_id}", ttl=600)
    async def get_user_profile(user_id: int): ...
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            client: Redis = RedisManager.get_client()
            
            # 构造带参数哈希的缓存键
            args_str = str(sorted(kwargs.items())) if kwargs else str(args)
            args_hash = hashlib.md5(args_str.encode()).hexdigest()[:8]
            cache_key = f"{key_prefix}:{args_hash}"
            
            # 尝试获取缓存
            cached_val = await client.get(cache_key)
            if cached_val:
                return json.loads(cached_val)
            
            # 未命中，执行原函数
            result = await func(*args, **kwargs)
            
            # 存入缓存（自动处理简单复杂对象的序列化）
            await client.setex(
                cache_key,
                ttl,
                json.dumps(result, default=str)
            )
            return result
        return wrapper
    return decorator

async def invalidate_cache(pattern: str):
    """清除匹配的缓存键（生产建议用SCAN替代KEYS）"""
    client = RedisManager.get_client()
    keys = await client.keys(pattern)
    if keys:
        await client.delete(*keys)

# services/user_service.py
from cache.decorators import cached, invalidate_cache
from redis_client import RedisManager

# 缓存用户资料15分钟
@cached("user:profile:{user_id}", ttl=900)
async def get_user_profile(user_id: int):
    # 模拟数据库查询
    return {"id": user_id, "name": f"用户{user_id}", "email": f"{user_id}@example.com"}

# 更新后清除缓存
async def update_user_profile(user_id: int, data: dict):
    # 模拟数据库更新
    print(f"更新用户{user_id}：{data}")
    await invalidate_cache(f"user:profile:{user_id}*")  # 清除所有相关键

# 缓存热门文章1小时，带随机抖动防雪崩
@cached("posts:hot", ttl=3600 + (id % 100) for id in range(1000))
async def get_hot_posts():
    # 模拟数据库聚合查询
    return [{"id": i, "title": f"热门文章{i}"} for i in range(10)]

# session/manager.py
import uuid
import json
from datetime import timedelta
from redis.asyncio import Redis
from redis_client import RedisManager

class RedisSessionManager:
    def __init__(self, prefix: str = "session:", expire: int = 86400*7):
        self.client: Redis = RedisManager.get_client()
        self.prefix = prefix
        self.expire = expire
    
    async def create(self, user_id: int, **extra) -> str:
        """创建新Session，返回session_id"""
        session_id = str(uuid.uuid4())
        key = self.prefix + session_id
        data = {"user_id": user_id, **extra}
        await self.client.setex(key, self.expire, json.dumps(data))
        return session_id
    
    async def get(self, session_id: str) -> dict | None:
        """获取并刷新Session"""
        key = self.prefix + session_id
        data = await self.client.get(key)
        if data:
            await self.client.expire(key, self.expire)  # 续期
            return json.loads(data)
        return None
    
    async def destroy(self, session_id: str) -> bool:
        """登出时销毁Session"""
        key = self.prefix + session_id
        return await self.client.delete(key) > 0

session_manager = RedisSessionManager()

from fastapi import Request, Cookie, HTTPException, Depends
from session.manager import session_manager

async def get_current_user(
    request: Request,
    session_id: str = Cookie(None)
):
    """从Cookie或Header获取Session，返回当前用户"""
    # 也支持Header携带（X-Session-ID）
    header_id = request.headers.get("X-Session-ID")
    current_id = header_id or session_id
    
    if not current_id:
        raise HTTPException(401, "请先登录")
    
    session = await session_manager.get(current_id)
    if not session:
        raise HTTPException(401, "Session已过期，请重新登录")
    
    return session

# 使用示例
@app.get("/profile")
async def my_profile(user: dict = Depends(get_current_user)):
    return {"profile": await get_user_profile(user["user_id"])}

@app.post("/logout")
async def logout(session_id: str = Cookie(None)):
    if session_id:
        await session_manager.destroy(session_id)
    return {"msg": "登出成功"}

# queue/list_queue.py
import json
from redis_client import RedisManager

LIST_QUEUE = "fastapi:tasks"

async def enqueue_task(task_type: str, **task_data):
    """入队即时任务"""
    client = RedisManager.get_client()
    await client.rpush(LIST_QUEUE, json.dumps({"type": task_type, "data": task_data}))

async def process_list_queue():
    """后台无限循环处理队列（单独进程/容器运行）"""
    client = RedisManager.get_client()
    while True:
        # 阻塞式弹出，无任务等待0秒（永久）
        result = await client.blpop(LIST_QUEUE, timeout=0)
        if result:
            _, task_json = result
            task = json.loads(task_json)
            print(f"处理任务：{task}")
            # 根据type分发处理
            # if task["type"] == "send_welcome": await send_email(task["data"])

# queue/delay_queue.py
import json
import time
from redis_client import RedisManager

DELAY_QUEUE = "fastapi:delay_tasks"
TEMP_QUEUE = "fastapi:delay_temp"

async def enqueue_delay_task(task_type: str, delay_seconds: int, **task_data):
    """入队延迟任务（例如10秒后发送验证码）"""
    client = RedisManager.get_client()
    score = time.time() + delay_seconds
    await client.zadd(DELAY_QUEUE, {json.dumps({"type": task_type, "data": task_data}): score})

async def process_delay_queue():
    """后台轮询处理延迟任务（建议1秒轮询一次）"""
    client = RedisManager.get_client()
    while True:
        now = time.time()
        # 获取所有到期的任务
        tasks = await client.zrangebyscore(DELAY_QUEUE, 0, now, withscores=False)
        if tasks:
            # 用事务/管道原子操作：删除原任务 + 入队即时处理
            pipe = client.pipeline()
            pipe.zrem(DELAY_QUEUE, *tasks)
            for task in tasks:
                pipe.rpush(LIST_QUEUE, task)
            await pipe.execute()
        time.sleep(1)