| 10 | from fastapi_cache.decorator import cache as original_cache |
| 11 | |
| 12 | def custom_key_builder( |
| 13 | func: Any, |
| 14 | namespace: str = "", |
| 15 | *, |
| 16 | args: Tuple[Any, ...] = (), |
| 17 | kwargs: Dict[str, Any], |
| 18 | cacheName: str, |
| 19 | keyExpression: Optional[str] = None, |
| 20 | ) -> str | list[str]: |
| 21 | try: |
| 22 | base_key = f"{namespace}:{cacheName}:" |
| 23 | |
| 24 | if keyExpression: |
| 25 | sig = signature(func) |
| 26 | bound_args = sig.bind_partial(*args, **kwargs) |
| 27 | bound_args.apply_defaults() |
| 28 | |
| 29 | # 支持args[0]格式 |
| 30 | if keyExpression.startswith("args["): |
| 31 | if match := re.match(r"args\[(\d+)\]", keyExpression): |
| 32 | index = int(match.group(1)) |
| 33 | value = bound_args.args[index] |
| 34 | if isinstance(value, list): |
| 35 | return [f"{base_key}{v}" for v in value] |
| 36 | return f"{base_key}{value}" |
| 37 | |
| 38 | # 支持属性路径格式 |
| 39 | parts = keyExpression.split('.') |
| 40 | if not bound_args.arguments.get(parts[0]): |
| 41 | return f"{base_key}{parts[0]}" |
| 42 | value = bound_args.arguments[parts[0]] |
| 43 | for part in parts[1:]: |
| 44 | value = getattr(value, part) |
| 45 | if isinstance(value, list): |
| 46 | return [f"{base_key}{v}" for v in value] |
| 47 | return f"{base_key}{value}" |
| 48 | |
| 49 | # 默认使用第一个参数作为key |
| 50 | return f"{base_key}{args[0] if args else 'default'}" |
| 51 | |
| 52 | except Exception as e: |
| 53 | SQLBotLogUtil.error(f"Key builder error: {str(e)}") |
| 54 | raise ValueError(f"Invalid cache key generation: {e}") from e |
| 55 | |
| 56 | def cache( |
| 57 | expire: int = 60 * 60 * 24, |