chongming-cache (0.2.1)
Published 2026-09-04 14:49:07 +08:00 by chakcy
Installation
pip install --index-url http://124.71.68.6:3000/api/packages/chongming/pypi/simple/ --extra-index-url https://pypi.org/simple chongming-cacheAbout this package
统一缓存封装：兼容 sqlite(本地) / redis(共享) 后端
chongming-cache
统一缓存封装，dict-like 接口，后端可切换 sqlite（本机）/ redis（跨主机），通过配置一行切换，业务代码零改动。
安装
uv pip install -e utils/cache
# 或作为依赖接入（与 chongming-database 等同约定）：
# [tool.uv.sources]
# chongming-cache = { git = "http://124.71.68.6:3000/chongming/chongming-utils-cache.git" }
快速开始
from chongming_cache import Cache
# 方式一：从配置创建（主项目 config.toml 的 [cache] 段）
cache = Cache.from_config(config["cache"])
# 方式二：显式参数
cache = Cache("sqlite", path="./cache_data/cache.db")
# cache = Cache("redis", redis_host="127.0.0.1", redis_port=6379, redis_prefix="chongming:")
# 历史后端名 "diskcache" 兼容映射为 sqlite
映射协议（与 dict / diskcache 用法一致）
cache["key"] = {"a": 1} # 写入（无 TTL）
if "key" in cache: # 存在性检查（含 TTL 判断）
value = cache["key"] # 读取，未命中抛 KeyError
del cache["key"] # 删除，未命中抛 KeyError
len(cache) # 条目数
for k in cache: ... # 迭代 key
扩展接口
cache.get("key", default=None) # 未命中返回 default
cache.set("key", value, expire=60) # 写入，TTL 秒（None = 永久）
cache.touch("key", expire=120) # 刷新 TTL，返回是否命中（未命中 False）
cache.delete("key") # 删除，返回是否命中
cache.clear() # 清空（redis 仅清本前缀命名空间）
cache.close() # 释放连接
with Cache.from_config(cfg) as cache: ... # 上下文管理器，退出自动 close
cache.client # 底层客户端逃生口（sqlite3.Connection / redis.Redis）
# 原子写（分布式锁原语）：仅当 key 不存在时写入，成功返回 True，已存在返回 False，不覆盖
ok = cache.backend.set_nx("leader", {"node": "gw-1"}, expire=30)
分布式排他锁 DistributedLock
基于 backend.set_nx（SET NX + TTL 租约）实现的跨进程/跨主机排他锁，用于集群协调（如保证全局只有一个 MQTT Bridge 消费）：
import asyncio
from chongming_cache import Cache, DistributedLock
async def main():
cache = Cache.from_config(config["cache"])
lock = DistributedLock(cache, "mqtt_bridge", ttl=30, renew_interval=10)
if not lock.acquire(): # 全局唯一持锁方返回 True，其余返回 False
print("其他实例已持有，跳过")
return
lock.start() # 后台自动续约（防宕机/长暂停导致锁过期被抢）
try:
... # 执行受保护的单例任务
finally:
await lock.stop() # 停止续约
lock.release() # 受保护释放
cache.close()
asyncio.run(main())
acquire()：SET NX，原子地仅在不存在时写入；key 带 TTL 租约，持有方宕机后到期自动让出，不会死锁。start()/stop()：后台续约任务；续约前校验 token，仅在自己仍持有时touch，锁丢失即停止续约。release()：受保护释放——仅当 token 仍匹配（仍是自己持有）才删除，绝不误删其他进程/主机抢到的锁。- 后端适用性：
redis真正跨主机共享；sqlite基于共享文件（SQLite 多进程安全），可用于单机多进程或共享卷场景，不可跨无共享文件系统的主机。
后端配置（config.toml [cache] 段）
[cache]
backend = "sqlite" # "sqlite"（默认，本机共享） | "redis"（跨主机）
path = "./cache_data/cache.db" # 兼容旧键 cache_store_path（指向目录时自动追加 cache.db）
# redis 后端专用（backend = "redis" 时生效，均有默认值）
# redis_host = "127.0.0.1"
# redis_port = 6379
# redis_db = 0
# redis_password = ""
# redis_prefix = "chongming:"
所有键均可缺省：旧配置（只有 cache_store_path）自动按 sqlite 后端工作。
双后端语义对齐
| 行为 | sqlite（本机） | redis（跨主机） |
|---|---|---|
| 值序列化 | JSON(orjson) | JSON(orjson)（与 Rust 进程插件 serde_json 互通） |
未命中 cache[key] |
KeyError |
KeyError |
set(..., expire=N) |
秒级 TTL，读时惰性淘汰 | SET EX N，到期即消失 |
touch(未命中) |
返回 False | 返回 False |
clear() |
清空本地缓存 | 仅清 redis_prefix 命名空间（不 flushdb） |
| 多进程安全 | 是（SQLite 文件 + WAL） | 是（服务端原子性） |
跨语言共享（Rust 插件 / sqlite）
sqlite 后端把值以 JSON 文本存进表 cache(k TEXT PRIMARY KEY, v TEXT NOT NULL, expire REAL)，
schema 与 Rust 进程插件（chongming-worker-rs，feature local-cache）完全一致——Python 与
Rust 读写同一份 .db 文件即可互通，无需 Redis。Rust 侧读侧不改变该 schema。
依赖
sqlite3— 本地共享后端（Python 标准库，零第三方依赖）redis— 远程后端（懒加载，仅选择 redis 后端时才 import）orjson— 全后端统一 JSON 序列化（缺失时回退标准库json）
Requirements
Requires Python: >=3.12
Details
2026-09-04 14:49:07 +08:00
Assets (2)
Versions (4)
View all
PyPI
8
60 KiB