litefs.cache package

class litefs.cache.CacheBackend[源代码]

基类:object

缓存后端类型

DATABASE = 'database'
MEMCACHE = 'memcache'
MEMORY = 'memory'
REDIS = 'redis'
TREE = 'tree'
class litefs.cache.CacheFactory[源代码]

基类:object

缓存工厂

根据配置创建不同类型的缓存实例

static create_cache(backend: str = 'memory', **kwargs) MemoryCache | TreeCache | RedisCache | DatabaseCache | MemcacheCache[源代码]

创建缓存实例

参数:
  • backend -- 缓存后端类型(memory, tree, redis, database, memcache)

  • **kwargs -- 缓存配置参数

返回:

缓存实例

抛出:
  • ValueError -- 不支持的缓存后端

  • ImportError -- Redis 或 Memcache 包未安装

static create_from_config(config) MemoryCache | TreeCache | RedisCache | DatabaseCache | MemcacheCache[源代码]

从配置对象创建缓存实例

参数:

config -- 配置对象,应包含 cache_backend 和相关配置

返回:

缓存实例

class litefs.cache.CacheManager[源代码]

基类:object

全局缓存管理器(单例模式)

确保缓存对象在应用生命周期内常驻内存,不会因为 Litefs 实例的 创建和销毁而丢失数据。

使用示例:

# 获取缓存实例(自动创建) cache = CacheManager.get_cache()

# 获取指定类型的缓存 session_cache = CacheManager.get_cache(

backend='memory', max_size=1000000, cache_key='sessions'

)

# 重置缓存(谨慎使用) CacheManager.reset_cache()

classmethod get_cache(backend: str = 'tree', cache_key: str | None = None, **kwargs) MemoryCache | TreeCache | RedisCache | DatabaseCache | MemcacheCache[源代码]

获取缓存实例(单例模式)

如果缓存实例不存在,则自动创建。同一 cache_key 的缓存实例 在整个应用生命周期内保持唯一。

参数:
  • backend -- 缓存后端类型

  • cache_key -- 缓存实例标识,None 使用默认缓存

  • **kwargs -- 缓存配置参数

返回:

缓存实例

classmethod get_file_cache(**kwargs)[源代码]

获取文件缓存实例

参数:

**kwargs -- 配置参数,支持 clean_period, expiration_time

返回:

TreeCache 实例

classmethod get_session_cache(**kwargs)[源代码]

获取会话缓存实例

参数:

**kwargs -- 配置参数,支持 max_size

返回:

MemoryCache 实例

classmethod has_cache(cache_key: str) bool[源代码]

检查缓存实例是否存在

参数:

cache_key -- 缓存标识

返回:

是否存在

classmethod list_caches() list[源代码]

获取所有缓存标识列表

返回:

缓存标识列表

classmethod reset_cache(cache_key: str | None = None) None[源代码]

重置缓存实例

谨慎使用!重置后缓存数据将丢失。

参数:

cache_key -- 要重置的缓存标识,None 重置所有缓存

class litefs.cache.DatabaseCache(db_path: str = ':memory:', table_name: str = 'cache', expiration_time: int = 3600, **kwargs)[源代码]

基类:object

数据库缓存实现

使用 SQLite 数据库作为缓存后端,提供持久化的缓存支持

clear() None[源代码]

清空所有缓存

close() None[源代码]

关闭数据库连接

delete(key: str) None[源代码]

从缓存删除值

参数:

key -- 缓存键

delete_many(keys: list) None[源代码]

批量删除键

参数:

keys -- 缓存键列表

delete_pattern(pattern: str) int[源代码]

删除匹配模式的键

参数:

pattern -- 键模式(支持 SQL LIKE 语法)

返回:

删除的键数量

exists(key: str) bool[源代码]

检查键是否存在

参数:

key -- 缓存键

返回:

键是否存在且未过期

expire(key: str, expiration: int) bool[源代码]

设置键的过期时间

参数:
  • key -- 缓存键

  • expiration -- 过期时间(秒)

返回:

是否设置成功

get(key: str) Any | None[源代码]

从缓存获取值

参数:

key -- 缓存键

返回:

缓存值,如果不存在或已过期则返回 None

get_many(keys: list) dict[源代码]

批量获取值

参数:

keys -- 缓存键列表

返回:

键值字典

put(key: str, val: Any, expiration: int | None = None) None[源代码]

存储值到缓存

参数:
  • key -- 缓存键

  • val -- 缓存值

  • expiration -- 过期时间(秒),如果为 None 则使用默认过期时间

set_many(mapping: dict, expiration: int | None = None) None[源代码]

批量存储值

参数:
  • mapping -- 键值字典

  • expiration -- 过期时间(秒),如果为 None 则使用默认过期时间

ttl(key: str) int[源代码]

获取键的剩余过期时间

参数:

key -- 缓存键

返回:

剩余过期时间(秒),如果键不存在或已过期则返回 -2

class litefs.cache.FileEventHandler(app)[源代码]

基类:FileSystemEventHandler

add_monitored_file(path)[源代码]

添加需要监控的文件

参数:

path -- 文件路径

on_created(event)[源代码]

Called when a file or directory is created.

参数:

event (DirCreatedEvent or FileCreatedEvent) -- Event representing file/directory creation.

on_deleted(event)[源代码]

Called when a file or directory is deleted.

参数:

event (DirDeletedEvent or FileDeletedEvent) -- Event representing file/directory deletion.

on_modified(event)[源代码]

Called when a file or directory is modified.

参数:

event (DirModifiedEvent or FileModifiedEvent) -- Event representing file/directory modification.

on_moved(event)[源代码]

Called when a file or a directory is moved or renamed.

参数:

event (DirMovedEvent or FileMovedEvent) -- Event representing file/directory movement.

class litefs.cache.FormCache(max_size: int = 1000, default_ttl: int = 300)[源代码]

基类:object

表单数据缓存

使用 LRU (Least Recently Used) 策略管理缓存 支持自动过期和容量限制

cleanup_expired() int[源代码]

清理过期的缓存项

返回:

清理的项数

clear() None[源代码]

清空所有缓存

delete(key: str) bool[源代码]

删除缓存项

参数:

key -- 缓存键

返回:

是否删除成功

get(key: str) Dict[str, Any] | None[源代码]

获取缓存的表单数据

参数:

key -- 缓存键

返回:

缓存的表单数据,如果不存在或已过期则返回 None

get_stats() Dict[str, Any][源代码]

获取缓存统计信息

返回:

缓存统计信息字典

set(key: str, value: Dict[str, Any], ttl: int | None = None) None[源代码]

设置缓存的表单数据

参数:
  • key -- 缓存键

  • value -- 表单数据

  • ttl -- 过期时间(秒),None 使用默认值

class litefs.cache.LiteFile(path, base, name, text, status_code=200)[源代码]

基类:object

handler(request)[源代码]
class litefs.cache.MemcacheCache(memcache_client=None, servers: list = ['localhost:11211'], key_prefix: str = 'litefs:', expiration_time: int = 3600, **kwargs)[源代码]

基类:object

Memcache 缓存实现

使用 Memcache 作为缓存后端,提供高性能的分布式缓存支持

clear() None[源代码]

清空所有缓存

注意:此方法仅删除带前缀的键,需要遍历所有键

close() None[源代码]

关闭 Memcache 连接

delete(key: str) None[源代码]

从缓存删除值

参数:

key -- 缓存键

delete_many(keys: list) None[源代码]

批量删除键

参数:

keys -- 缓存键列表

delete_pattern(pattern: str) int[源代码]

删除匹配模式的键

注意:Memcache 不支持模式匹配,此方法仅返回 0

参数:

pattern -- 键模式

返回:

删除的键数量(始终为 0)

exists(key: str) bool[源代码]

检查键是否存在

参数:

key -- 缓存键

返回:

键是否存在

expire(key: str, expiration: int) bool[源代码]

设置键的过期时间

参数:
  • key -- 缓存键

  • expiration -- 过期时间(秒)

返回:

是否设置成功

get(key: str) Any | None[源代码]

从缓存获取值

参数:

key -- 缓存键

返回:

缓存值,如果不存在则返回 None

get_many(keys: list) dict[源代码]

批量获取值

参数:

keys -- 缓存键列表

返回:

键值字典

put(key: str, val: Any, expiration: int | None = None) None[源代码]

存储值到缓存

参数:
  • key -- 缓存键

  • val -- 缓存值

  • expiration -- 过期时间(秒),如果为 None 则使用默认过期时间

set_many(mapping: dict, expiration: int | None = None) None[源代码]

批量存储值

参数:
  • mapping -- 键值字典

  • expiration -- 过期时间(秒),如果为 None 则使用默认过期时间

ttl(key: str) int[源代码]

获取键的剩余过期时间

注意:Memcache 不支持 TTL 查询,此方法返回 -1

参数:

key -- 缓存键

返回:

剩余过期时间(秒),如果键不存在则返回 -2,否则返回 -1(不支持查询)

class litefs.cache.MemoryCache(max_size=10000)[源代码]

基类:object

delete(key)[源代码]
get(key)[源代码]
put(key, val)[源代码]
class litefs.cache.RedisCache(redis_client=None, host: str = 'localhost', port: int = 6379, db: int = 0, password: str | None = None, key_prefix: str = 'litefs:', expiration_time: int = 3600, **kwargs)[源代码]

基类:object

Redis 缓存实现

使用 Redis 作为缓存后端,提供高性能的分布式缓存支持

clear() None[源代码]

清空所有缓存

注意:这会删除所有带前缀的键

close() None[源代码]

关闭 Redis 连接

delete(key: str) None[源代码]

从缓存删除值

参数:

key -- 缓存键

delete_many(keys: list) None[源代码]

批量删除键

参数:

keys -- 缓存键列表

delete_pattern(pattern: str) int[源代码]

删除匹配模式的键

参数:

pattern -- 键模式

返回:

删除的键数量

exists(key: str) bool[源代码]

检查键是否存在

参数:

key -- 缓存键

返回:

键是否存在

expire(key: str, expiration: int) bool[源代码]

设置键的过期时间

参数:
  • key -- 缓存键

  • expiration -- 过期时间(秒)

返回:

是否设置成功

get(key: str) Any | None[源代码]

从缓存获取值

参数:

key -- 缓存键

返回:

缓存值,如果不存在则返回 None

get_many(keys: list) dict[源代码]

批量获取值

参数:

keys -- 缓存键列表

返回:

键值字典

put(key: str, val: Any, expiration: int | None = None) None[源代码]

存储值到缓存

参数:
  • key -- 缓存键

  • val -- 缓存值

  • expiration -- 过期时间(秒),如果为 None 则使用默认过期时间

set_many(mapping: dict, expiration: int | None = None) None[源代码]

批量存储值

参数:
  • mapping -- 键值字典

  • expiration -- 过期时间(秒),如果为 None 则使用默认过期时间

ttl(key: str) int[源代码]

获取键的剩余过期时间

参数:

key -- 缓存键

返回:

剩余过期时间(秒),如果键不存在则返回 -2

class litefs.cache.TreeCache(clean_period=60, expiration_time=3600)[源代码]

基类:object

auto_clean()[源代码]

优化的清理机制

避免每次清理都查询数据库,只在清理周期到达时执行清理

delete(key)[源代码]
get(key)[源代码]
put(key, val)[源代码]
litefs.cache.get_global_cache(backend: str = 'tree', **kwargs) MemoryCache | TreeCache | RedisCache | DatabaseCache | MemcacheCache[源代码]

获取全局缓存实例的便捷函数

参数:
  • backend -- 缓存后端类型

  • **kwargs -- 缓存配置参数

返回:

缓存实例

Submodules

litefs.cache.cache module

class litefs.cache.cache.FileEventHandler(app)[源代码]

基类:FileSystemEventHandler

add_monitored_file(path)[源代码]

添加需要监控的文件

参数:

path -- 文件路径

on_created(event)[源代码]

Called when a file or directory is created.

参数:

event (DirCreatedEvent or FileCreatedEvent) -- Event representing file/directory creation.

on_deleted(event)[源代码]

Called when a file or directory is deleted.

参数:

event (DirDeletedEvent or FileDeletedEvent) -- Event representing file/directory deletion.

on_modified(event)[源代码]

Called when a file or directory is modified.

参数:

event (DirModifiedEvent or FileModifiedEvent) -- Event representing file/directory modification.

on_moved(event)[源代码]

Called when a file or a directory is moved or renamed.

参数:

event (DirMovedEvent or FileMovedEvent) -- Event representing file/directory movement.

class litefs.cache.cache.LiteFile(path, base, name, text, status_code=200)[源代码]

基类:object

handler(request)[源代码]
class litefs.cache.cache.MemoryCache(max_size=10000)[源代码]

基类:object

delete(key)[源代码]
get(key)[源代码]
put(key, val)[源代码]
class litefs.cache.cache.TreeCache(clean_period=60, expiration_time=3600)[源代码]

基类:object

auto_clean()[源代码]

优化的清理机制

避免每次清理都查询数据库,只在清理周期到达时执行清理

delete(key)[源代码]
get(key)[源代码]
put(key, val)[源代码]