FQuant Document

Appearance

Cache - API参考 ​

阅读路径 ​

🔵 开发者：README → api → usage → concepts → examples

类 ​

LocalCache ​

位置： Cache/local_cache.py#L26

描述： 本地内存缓存实现

python
from FQBase.Cache import LocalCache

cache = LocalCache(name="my_cache", ttl=3600, singleton=True)
cache.set("key", "value")
value = cache.get("key")

参数 ​

参数类型必填默认值描述
namestr是-缓存名称
ttlint否0默认过期时间（秒），0 表示永不过期
singletonbool否False是否单例模式

方法 ​

set ​
python
cache.set(key, value, ttl=None)

参数：

参数类型必填默认值描述
keystr是-缓存键
valueAny是-缓存值
ttlint否None过期时间（秒）
get ​
python
value = cache.get(key, default=None)
delete ​
python
cache.delete(key)
clear ​
python
cache.clear()
keys ​
python
keys = cache.keys(pattern="*")
get_many ​
python
result = cache.get_many(keys)

参数：

参数类型必填默认值描述
keysList[str]是-缓存键列表

返回： Dict[str, Any] - 存在的键值对

set_many ​
python
cache.set_many(mapping, ttl=None)

参数：

参数类型必填默认值描述
mappingDict[str, Any]是-键值对字典
ttlint否None过期时间（秒）
delete_many ​
python
cache.delete_many(keys)

参数：

参数类型必填默认值描述
keysList[str]是-缓存键列表

返回： bool - 是否至少删除了一个键

exists ​
python
result = cache.exists(key)

参数：

参数类型必填默认值描述
keystr是-缓存键

返回： bool - 键是否存在且未过期

ttl ​
python
remaining = cache.ttl(key)

参数：

参数类型必填默认值描述
keystr是-缓存键

返回： int - 剩余生存时间（秒），-1 表示永久，-2 表示不存在

expire ​
python
cache.expire(key, ttl)

参数：

参数类型必填默认值描述
keystr是-缓存键
ttlint是-过期时间（秒）

返回： bool - 是否设置成功

cleanup_expired ​
python
count = cache.cleanup_expired()

返回： int - 清理的缓存项数量

stats ​
python
stats = cache.stats

返回： Dict[str, Union[str, int, float]] - 缓存统计信息，包含 name, maxsize, ttl, eviction, size, hits, misses, hit_rate

has_key ​
python
result = cache.has_key(key)

参数：

参数类型必填默认值描述
keystr是-缓存键

返回： bool - 键是否存在且未过期（不触发过期删除）

RedisCacheAdapter ​

位置： Cache/redis_adapter.py#L198

描述： Redis 缓存适配器

python
from FQBase.Cache import RedisCacheAdapter

adapter = RedisCacheAdapter()
adapter.set("key", "value", ttl=3600)

参数 ​

参数类型必填默认值描述
configRedisConfigProtocol否NoneRedis 配置

方法 ​

ping ​
python
result = adapter.ping()

返回： bool - Redis 连接是否正常

get ​
python
value = adapter.get(key, default=None)
set ​
python
adapter.set(key, value, ttl=None)
delete_many ​
python
adapter.delete_many(keys)

MongoCacheAdapter ​

位置： Cache/mongo_adapter.py#L22

描述： MongoDB 缓存适配器

python
from FQBase.Cache import MongoCacheAdapter

adapter = MongoCacheAdapter()
adapter.set("key", "value", ttl=3600)

函数 ​

create_cache ​

位置： Cache/__init__.py#L83

python
from FQBase.Cache import create_cache

cache = create_cache(config=None)

描述： 根据配置创建缓存适配器（工厂方法）

参数：

参数类型必填默认值描述
configCacheConfigProtocol否None缓存配置

返回： CacheInterface - 缓存适配器实例

init_cache_adapter ​

位置： Cache/CacheAdapters.py#L217

python
from FQBase.Cache import init_cache_adapter

init_cache_adapter()

描述： 从环境变量初始化全局缓存适配器，失败时降级到 Memory

get_cache_adapter ​

位置： Cache/CacheAdapters.py#L256

python
from FQBase.Cache import get_cache_adapter

adapter = get_cache_adapter()

描述： 获取全局缓存适配器，首次调用时自动初始化

返回： Optional[RedisCacheAdapter]

set_cache_adapter ​

位置： Cache/CacheAdapters.py#L243

python
from FQBase.Cache import set_cache_adapter

set_cache_adapter(adapter)

描述： 设置全局缓存适配器

invalidate_cache ​

位置： Cache/CacheAdapters.py#L267

python
from FQBase.Cache import invalidate_cache

invalidate_cache(key_pattern="user:*")

描述： 使缓存失效

参数：

参数类型必填默认值描述
key_patternstr否*缓存键模式

redis_cache ​

位置： Cache/CacheAdapters.py#L132

python
from FQBase.Cache import redis_cache

@redis_cache(ttl=300, key_prefix="func")
def my_func(arg):
    return expensive_operation(arg)

描述： Redis 缓存装饰器，支持异步函数

参数：

参数类型必填默认值描述
ttlint否300缓存过期时间（秒）
key_prefixstr否""缓存键前缀
key_ttl_funcCallable否None动态 TTL 函数

local_cache ​

位置： Cache/local_cache.py#L488

python
from FQBase.Cache import local_cache

@local_cache(ttl=300, maxsize=128)
def cached_func():
    return result

描述： 本地内存缓存装饰器，基于 LocalCache 实现

参数：

参数类型必填默认值描述
maxsizeint否128最大缓存条目数
ttlint否0过期时间（秒），0 表示不过期
key_ttl_funcCallable否None动态 TTL 函数

附加属性：

属性类型描述
cache_clearCallable清除缓存
cache_statsDict缓存统计信息

init_cache_from_env ​

位置： Cache/__init__.py#L111

python
from FQBase.Cache import init_cache_from_env

adapter = init_cache_from_env()

描述： 从环境变量初始化全局缓存（推荐入口）

返回： CacheInterface - 缓存适配器实例

register_cache_adapter ​

位置： Cache/__init__.py#L57

python
from FQBase.Cache import register_cache_adapter

register_cache_adapter('custom', lambda cfg: CustomAdapter(cfg))

描述： 注册自定义缓存适配器工厂

参数：

参数类型必填默认值描述
cache_typestr是-缓存类型标识
factory_funcCallable是-工厂函数，接收 config 返回适配器实例

示例：

python
from FQBase.Cache import register_cache_adapter, create_cache

class CustomCacheAdapter:
    def __init__(self, config):
        self._config = config

    def get(self, key):
        return self._backend.get(key)

    def set(self, key, value, ttl=0):
        return self._backend.set(key, value, ttl)

register_cache_adapter('custom', lambda cfg: CustomCacheAdapter(cfg))
cache = create_cache(CustomCacheConfig(cache_type='custom'))

异常 ​

异常描述触发条件解决方案
CacheError缓存基类异常缓存操作失败检查缓存服务状态
CacheConnectionError连接错误Redis/MongoDB 连接失败检查服务配置
CacheTimeoutError超时错误缓存操作超时增加超时时间
CacheSerializationError序列化错误对象无法序列化检查数据类型
CacheKeyError键错误缓存键无效检查键格式
CacheOperationError操作错误缓存操作失败查看具体错误信息

相关文档 ​