Python FastAPI 实现函数结果缓存

在 FastAPI 中实现函数结果缓存，可以使用 cachetools 库或者 FastAPI 内置的 Depends 结合缓存机制。下面我将展示两种实现方式：

方法一：使用 cachetools 库

首先安装依赖：

pip install cachetools fastapi

然后实现缓存：

from fastapi import FastAPI
from cachetools import cached, TTLCache
from datetime import datetime, timedelta

app = FastAPI()

# 创建缓存，最大100个条目，每个条目存活1800秒（30分钟）
cache = TTLCache(maxsize=100, ttl=1800)

@app.get("/expensive-operation/{param}")
@cached(cache)
async def expensive_operation(param: str):
    """
    这个函数会被缓存，30分钟内相同参数的调用会直接返回缓存结果
    """
    # 模拟耗时操作
    print(f"执行耗时操作，参数: {param}")
    result = {"data": f"处理结果 for {param}", "timestamp": datetime.now().isoformat()}
    return result

方法二：自定义缓存装饰器

如果你需要更多控制，可以自己实现一个缓存装饰器：

from fastapi import FastAPI
from functools import wraps
from datetime import datetime, timedelta
from typing import Any, Callable, Dict, Tuple, Optional

app = FastAPI()

# 缓存字典
function_cache: Dict[Tuple, Dict[str, Any]] = {}

def cache_result(ttl: int = 1800):
    """
    缓存装饰器，默认30分钟(1800秒)有效期
    """
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 创建缓存键（基于参数）
            cache_key = (func.__name__, args, tuple(sorted(kwargs.items())))
            
            # 检查缓存中是否有结果
            now = datetime.now()
            if cache_key in function_cache:
                cached_result, cached_time = function_cache[cache_key]
                if (now - cached_time).total_seconds() <= ttl:
                    print(f"返回缓存结果 for {cache_key}")
                    return cached_result
            
            # 执行函数并缓存结果
            print(f"执行函数 for {cache_key}")
            result = await func(*args, **kwargs)
            function_cache[cache_key] = (result, now)
            return result
        return wrapper
    return decorator

@app.get("/another-expensive-operation/{param}")
@cache_result()  # 默认30分钟
async def another_expensive_operation(param: str):
    """
    使用自定义缓存装饰器的函数
    """
    # 模拟耗时操作
    result = {"data": f"另一个处理结果 for {param}", "timestamp": datetime.now().isoformat()}
    return result

方法三：使用 FastAPI 的 Depends 和状态

对于更复杂的应用，可以结合 FastAPI 的依赖注入系统：

from fastapi import FastAPI, Depends, HTTPException
from datetime import datetime, timedelta
from typing import Dict, Optional

app = FastAPI()

class CacheManager:
    def __init__(self):
        self._cache: Dict[str, Dict[str, Any]] = {}
    
    async def get_or_set(self, key: str, func: Callable, *args, **kwargs) -> Any:
        # 检查缓存
        if key in self._cache:
            cached_data, expires_at = self._cache[key]
            if datetime.now() < expires_at:
                return cached_data
        
        # 执行函数
        result = await func(*args, **kwargs)
        
        # 设置缓存（30分钟）
        expires_at = datetime.now() + timedelta(minutes=30)
        self._cache[key] = {
            "data": result,
            "expires_at": expires_at
        }
        
        return result

cache_manager = CacheManager()

def get_cache_key(param: str) -> str:
    return f"expensive_op_{param}"

@app.get("/dependent-operation/{param}")
async def dependent_operation(param: str, cache: CacheManager = Depends()):
    cache_key = get_cache_key(param)
    
    def fetch_data():
        # 模拟耗时操作
        print(f"执行耗时操作，参数: {param}")
        return {"data": f"依赖注入缓存结果 for {param}", "timestamp": datetime.now().isoformat()}
    
    result = await cache.get_or_set(cache_key, fetch_data)
    return result

注意事项

1. 缓存键：确保缓存键能唯一标识函数调用，通常基于函数名和参数
2. 线程安全：如果使用多线程，确保缓存实现是线程安全的
3. 内存限制：对于生产环境，考虑使用 Redis 等分布式缓存替代内存缓存
4. 敏感数据：不要缓存敏感数据或包含用户特定信息的数据
5. 缓存失效：考虑如何处理数据更新时的缓存失效问题

对于生产环境，建议使用专业的缓存系统如 Redis，可以通过 aioredis 或 redis 库实现分布式缓存。