xiaoe-config-manager
多进程安全的 JSON 配置文件管理器,纯 Python 标准库实现,零外部依赖。
标准接口是 get / set / reset(2.1.0 起)。老名字 configget / configset /
configreset 仍然可用,它们在本层反向转发到新名字 —— 子类只需要实现新的三个。
特性
- 标准三件套 —
get/set/reset是接口;configget/configset/configreset是兼容壳,转发到它们 - 统一接口 —
ConfigApi(ABC)定义get/set/reset+ 订阅on/off/value_changed/off_value_changed(共 10 个公开成员),文件版 / 内存版 / 子进程版都是它的实现 - 声明式配置 —
ConfigKeysBase(纯声明)+ConfigDeclareMixin(点操作属性)+ConfigItemProxy,键名与默认值只写一处 - 线程安全 — 内置线程锁,单进程多线程环境下安全读写
- 多进程安全 — 通过
multiprocessing.Queue双向管道实现主进程与子进程之间的配置自动同步 - 变更通知自动 — 本进程写当场派发;跨进程写按逐键 diff 派发;自己的回声被 diff 吞掉,不会双发
- 数据安全 — 配置文件损坏时原样抛异常、绝不覆盖;落盘是原子写(写
.tmp再os.replace) - 点操作 API —
cfg.app_name.get() / .set(v) / .reset(),配合ConfigItemProxy[T]提供完整静态类型检查 - 类型注解友好 —
default()自动推断,ConfigItemProxy[T]显式声明,IDE 补全与类型检查全程可用 - 默认配置 — 支持静态默认字典和动态回调两种方式,配置项缺失时自动补齐
- 配置重置 — 支持单个 key 或全部配置一键恢复默认值
- 默认值热更新 — 当默认配置变化时,自动识别用户未修改的项并更新为新默认值
- 内存版 — 和文件版同一种创建方式:定义类(
class AppMem(ConfigMemory, AppKeys)→AppMem()),同样是声明默认值 + 点操作,只是不落盘;Config.memory({...})作为兼容 / 快速路径保留 - 纯标准库 — 无任何第三方依赖,仅使用
json、threading、multiprocessing等标准库
运行要求
| 项 | 要求 |
|---|---|
| Python | >= 3.10 |
| 运行时依赖 | 零依赖(纯标准库) |
| 跑测试 | pip install -e ".[dev]"(pytest 在 [dev] extra 里) |
安装
pip install xiaoe-config-manager
快速开始
定义配置类
单进程场景直接给 Config 接一份声明(或把默认值写在 Config 子类自己的类体里)。
from xiaoe_config_manager import Config, ConfigKeysBase, default
class AppKeys(ConfigKeysBase):
app_name = default("myapp")
timeout = default(30)
debug = default(False)
id_list = default(None, list)
class AppConfig(Config, AppKeys):
...
也可以直接
class AppConfig(Config)把默认值写在Config的子类类体里 —— 实现类自己的类体同样会被扫描。拆出AppKeys是为了多进程场景能两边共用一份声明, 见多进程使用。
使用
cfg = AppConfig(file_name="app_config")
# 点操作(推荐)
cfg.app_name.get() # "myapp"
cfg.timeout.set(60) # 修改并写文件
cfg.timeout.get() # 60
cfg.timeout.reset() # 恢复默认
cfg.timeout.get() # 30
# 标准三件套
cfg.set("timeout", 60)
cfg.get("timeout") # 60
cfg.get("all") # 返回全部配置字典
cfg.reset("all") # 恢复全部
# 老接口(等价,转发到上面三个)
cfg.configset("timeout", 60)
cfg.configget("timeout") # 60
cfg.configreset("all")
四种 API 风格对照
| 风格 | 写 | 读 | 重置 |
|---|---|---|---|
| 点操作(推荐) | cfg.timeout.set(60) |
cfg.timeout.get() |
cfg.timeout.reset() |
| 标准接口 | cfg.set("timeout", 60) |
cfg.get("timeout") |
cfg.reset("timeout") |
| 老接口(兼容壳) | cfg.configset("timeout", 60) |
cfg.configget("timeout") |
cfg.configreset("timeout") |
| 显式代理 | cfg.item("timeout").set(60) |
cfg.item("timeout").get() |
cfg.item("timeout").reset() |
四者操作同一份底层数据,可以混用。get("all") 是"整份配置"的统一入口。
最后一行是未声明的键(或撞了保留字的键)的入口 —— 声明的键直接 cfg.键名 就够了。
谁实现谁:子类实现的是
get/set/reset;configget/configset/configreset由ConfigApi提供,内部就是return self.get(key)这类一行转发。 2.0.x 及以前是反的,升级时注意不要把覆写点写错地方。
声明式配置
三样东西各管一段:
| 名字 | 职责 |
|---|---|
ConfigKeysBase |
纯声明基类。子类只放「键名 = 默认值」,__init_subclass__ 算出 DEFAULTS |
ConfigDeclareMixin |
在 ConfigKeysBase 之上,把每个键做成实例上的 property,cfg.speed 返回代理。ConfigBase 继承它 |
ConfigItemProxy[T] |
点操作代理,只转发 owner.get/set/reset/on/off |
default(value) / default(None, type_) |
声明默认值,运行时原样返回传入值,只为类型推断与可读性存在 |
一份声明可以多处消费:
from xiaoe_config_manager import Config, ConfigKeysBase, default
class AppKeys(ConfigKeysBase):
app_name = default("myapp")
AppKeys.DEFAULTS # {'app_name': 'myapp'}
Config("app", default_config=AppKeys.DEFAULTS) # 非声明式消费(照样能用)
class AppConfig(Config, AppKeys): ... # 声明式消费(顺带有点操作)
扫描规则
- 只扫
ConfigKeysBase的子类,以及实现类自己的类体;后端类 (Config/ConfigMemory/ 管道 / Qt 桥)的类字典一律不扫。 - 默认值必须是
str / int / float / bool / list / dict / tuple / None之一 (即DATA_TYPES)—— 函数、property 之类不是配置项,会被忽略。JSON里没有tuple,写进去读回来会变成list。 - 键名命中保留字当场
TypeError(不静默):先查显式清单(config/fname/queues/default_config/value_changed/item/keys/instance/memory/persist/DEFAULTS等),再动态查实现类 MRO 上已有的属性名 (off_value_changed/on/off/get/set/reset这些方法名就是这么被拦下的, 所以它们同样不能当配置键)。 - 一条继承链上重复声明同一个键 →
TypeError(class Child(AppKeys)里又写了一遍app_name);兄弟基类同名不拦,按 Python 的 MRO 走 —— 左边的赢。
class ParentKeys(ConfigKeysBase):
app_name = default("p")
class ChildKeys(ParentKeys):
app_name = default("c") # TypeError:重复声明了继承链上已有的配置键
撞了保留字的键(或压根没声明的键)想访问,用显式入口:cfg.item("原名")。
类型注解
模块所有点操作 API 都支持静态类型检查。下面用几个例子分别说明。
方式一:default() 自动推断
default(30) 类型检查器会看到 ConfigItemProxy[int],链式调用 .get() 返回 int。
class AppKeys(ConfigKeysBase):
timeout = default(30) # ConfigItemProxy[int]
app_name = default("myapp") # ConfigItemProxy[str]
debug = default(False) # ConfigItemProxy[bool]
class AppConfig(Config, AppKeys):
...
cfg = AppConfig(file_name="app")
n: int = cfg.timeout.get() # ✅ 类型检查通过
cfg.timeout.set(60) # ✅ int,通过
cfg.timeout.set("oops") # ❌ 类型检查报错:期望 int,收到 str
方式二:default(None, type) 声明可空类型
第一个参数为 None 时,第二个参数提供类型信息,推断结果为 ConfigItemProxy[T | None]。
class AppKeys(ConfigKeysBase):
id_list = default(None, list) # ConfigItemProxy[list | None]
mapping = default(None, dict) # ConfigItemProxy[dict | None]
name = default(None, str) # ConfigItemProxy[str | None]
class AppConfig(Config, AppKeys):
...
cfg = AppConfig(file_name="app")
# IDE 会提示 get() 可能是 None,需要判空
value = cfg.id_list.get() # Optional[list]
if value is not None:
value.append(1) # ✅ 收窄为 list
cfg.id_list.set(None) # ✅ None 允许
cfg.id_list.set([1, 2, 3]) # ✅ list 也允许
cfg.id_list.set("oops") # ❌ 报错
方式三:显式 ConfigItemProxy[T] 注解
需要更精确的类型(例如带元素类型的列表)时,直接写注解。
from xiaoe_config_manager import Config, ConfigItemProxy, ConfigKeysBase, default
from typing import Optional
class AppKeys(ConfigKeysBase):
# 更精确的元素类型
ports: ConfigItemProxy[list[int]] = default(None, list)
names: ConfigItemProxy[list[str]] = default(None, list)
# 可空 + 精确元素类型
tags: ConfigItemProxy[Optional[list[str]]] = default(None, list)
# 字典
routes: ConfigItemProxy[dict[str, str]] = default(None, dict)
class AppConfig(Config, AppKeys):
...
cfg = AppConfig(file_name="app")
ports = cfg.ports.get() # Optional[list[int]]
方式四:省略 default() 直接写裸值
也可以只写裸值,运行时行为完全一样,但 IDE 只看到原始类型,点操作的 .get() 不会被静态检查。
class AppKeys(ConfigKeysBase):
timeout = 30 # IDE 看到的是 int,而非 ConfigItemProxy[int]
app_name = "myapp" # IDE 看到的是 str
class AppConfig(Config, AppKeys):
...
cfg = AppConfig(file_name="app")
cfg.timeout.get() # 运行时正常,但没有类型检查与补全
cfg.timeout.set("oops") # 运行时才报错
推荐始终使用
default()或ConfigItemProxy[T],让 IDE 全程有提示。
IDE 补全效果
cfg.timeout. # 补全:get / set / reset / on / off
cfg.app_name.s # 补全:set
cfg.timeout.get() # 返回类型:int
数据安全
这一节的三条是故障链上的三道闸,都别绕过。
1. 坏文件报错,绝不覆盖
Config(...) 捕获的只有 FileNotFoundError(第一次运行还没有文件,此时兜底成空配置)。
文件损坏、编码错、没权限 —— 一律原样抛异常:
# app.json 里是 "{ 这不是 JSON "
Config("app", default_config={"a": 1})
# json.JSONDecodeError: Expecting property name enclosed in double quotes: ...
抛异常时文件一个字节都不会变,也不会留下 .tmp。
为什么这么严:以前这里把坏文件吞成空配置,紧接着
__init__用默认值补缺并_save()落盘 —— 用户原来的配置永久消失,且没有任何提示。报错比静默丢数据好。
2. 落盘是原子写
_save() 先写 <file>.json.tmp,json.dump 成功之后才 os.replace(tmp, path):
def _save(self):
tmp_path = f"{self.path}.tmp"
with open(tmp_path, 'w', encoding='UTF-8') as f:
json.dump(self.config, f, indent=4, ensure_ascii=False)
os.replace(tmp_path, self.path)
直接以 'w' 打开原文件写,会在序列化失败时把文件截断成半截
(不可序列化的类型、磁盘写满),下次启动就是坏配置 —— 那正是上面那一步的起点。
现在的形式下,任何一步失败,原文件都还是完整的旧内容。
配套的一条:DATA_TYPES 故意不含 set。set 过得了声明校验,却会让
json.dump 在文件被截断之后失败 —— 就是从这一步开始的。
3. get('all') 会重读文件并替换内存副本
这是有意的副作用(见 API 参考):文件版 get('all') 从磁盘重读,
把 self.config 换成文件里的内容。好处是别人刚写进去的值能看到;代价是本进程
未落盘的中间状态会被覆盖。要读单键就用 get(key),它只读内存副本。
多进程使用
多进程场景下,主进程持有 Config 单例,每个子进程在主进程侧对应一个
ConfigInMainProcessPipe。子进程通过 ConfigInChildProcessPipe 与主进程通信。
由于 ConfigInChildProcessPipe 不是 Config 的子类,主进程和子进程无法从同一个
Config 子类继承。此时把配置声明单独抽成一个纯声明类(例如 AppKeys),
主进程与子进程分别继承它,两边共享同一套键名与类型信息。
共享声明 + 主进程 + 子进程
import multiprocessing as mp
import time
from xiaoe_config_manager import (
Config, ConfigInMainProcessPipe, ConfigInChildProcessPipe,
ConfigKeysBase, default,
)
# ── 1. 共享声明:只放「键名 = 默认值」──
class AppKeys(ConfigKeysBase):
workers = default(4)
worker_status = default(None, str)
master_status = default(None, str)
# ── 2. 主进程:Config + 同一份声明 ──
class AppConfig(Config, AppKeys):
...
# ── 3. 子进程:ConfigInChildProcessPipe + 同一份声明 ──
def worker(set_q, get_q):
class AppConfigChild(ConfigInChildProcessPipe, AppKeys):
...
cfg = AppConfigChild(set_q, get_q) # 构造时就同步到第一份配置
print("子进程读到 workers =", cfg.workers.get()) # 4
# 订阅:单键 / 所有键。跨进程的变更会自动派发到这里。
cfg.worker_status.on(lambda v: print("子进程自己写的 worker_status =", v))
cfg.value_changed(lambda k, v: print("子进程收到变更:", k, "=", v))
cfg.worker_status.set("running") # 乐观写:本地立即生效,消息发给主进程
# 等主进程那一发广播(真实程序里不用这样等:Qt 事件循环会替你跑)
for _ in range(100):
if cfg.master_status.get() == "ok":
break
time.sleep(0.05)
print("子进程看到 master_status =", cfg.master_status.get())
cfg.close()
if __name__ == "__main__":
cfg = AppConfig(file_name="my_config")
# 主进程的订阅也是自动的:子进程写进来就派发
cfg.on("worker_status", lambda v: print("主进程收到 worker_status =", v))
main_pipe = ConfigInMainProcessPipe(cfg) # 每个子进程一个
set_q, get_q = main_pipe.get_queues()
p = mp.Process(target=worker, args=(set_q, get_q))
p.start()
# 等子进程把 worker_status 写进来(真实程序里靠上面的订阅回调,不用轮询)
for _ in range(100):
if cfg.worker_status.get() == "running":
break
time.sleep(0.05)
cfg.master_status.set("ok") # 主进程写:当场派发,并广播给所有子进程
p.join()
print("主进程最终配置 =", cfg.get("all"))
main_pipe.close()
输出:
子进程读到 workers = 4
子进程自己写的 worker_status = running
子进程收到变更: worker_status = running
子进程收到变更: master_status = ok
子进程看到 master_status = ok
主进程收到 worker_status = running
主进程最终配置 = {'workers': 4, 'worker_status': 'running', 'master_status': 'ok'}
为什么需要 AppKeys
ConfigDeclareMixin在每一个继承它的子类上单独生成点操作属性- 主进程的
AppConfig与子进程的AppConfigChild是两个不同的子类,各自需要一份声明 - 把声明抽到
AppKeys,一次编写两处继承,键名、默认值、类型注解三方面都不会漂移 - 单进程场景只有一个类,直接
class AppConfig(Config, AppKeys)就够了,不必刻意拆
通知是自动的
不需要手工接线 —— on_remote_change / off_remote_change 已经删除,
notify_change(ui 侧)也删了。直接 cfg.on(key, cb) / cfg.value_changed(cb) 就行
(取消订阅分别是 cfg.off(key, cb) / cfg.off_value_changed(cb)):
| 场景 | 派发方式 |
|---|---|
本进程 set |
当场派发(set 返回时值已生效) |
本进程 reset(文件版 / 内存版) |
当场派发新值 |
本进程 reset(子进程管道) |
悲观:当场那一发可能还是旧值,广播回来时再按 diff 派发正确的新值 |
| 别的进程改了 → 广播回来 | 按逐键 diff 派发,只报真的变了的键 |
| 子进程写完、主进程广播回来的回声 | 被 diff 吞掉(before == after),不会双发 |
| 主进程处理子进程的写入 | 走 Config.set,派发点只有 ConfigBase._dispatch 一处,也不会双发 |
架构说明
子进程 cfg.set(v) ──→ set_queue ──→ 主进程 _loop() ──→ Config.set()
│
▼
config_get_queue(整份 config)
│
▼
子进程 _update_config_loop()
替换本地缓存 → diff_changes() → _dispatch()
- 子进程的
set()不直接写文件,通过队列把请求交给主进程;reset()同样只发消息 - 主进程统一处理所有写操作和文件 I/O,修改后通过广播队列通知所有子进程
- 子进程的
get()从本地缓存读取,零跨进程开销 - 每个子进程需要独立的一对队列(一个
ConfigInMainProcessPipe+ 一个ConfigInChildProcessPipe)
内存版
内存版和文件版用同一种方式创建 —— 定义类:继承 ConfigMemory 并把声明类接上;
不传参数就用声明的默认值,传一个 dict 就盖在上面。区别只有 fname is None(不落盘)。
from xiaoe_config_manager import Config, ConfigKeysBase, ConfigMemory, default
class AppKeys(ConfigKeysBase):
enable = default(False)
speed = default(50)
class AppMemory(ConfigMemory, AppKeys): # 内存版:类 + 不传
...
mem = AppMemory() # 声明里的默认值
mem.speed.get() # 50
mem.speed.set(60) # 只改内存,没有文件
mem.speed.get() # 60
mem.speed.reset() # 回到 50
mem2 = AppMemory({"speed": 99}) # 类 + 一份 dict:盖在声明默认值上面,同时是 reset 目标
mem2.speed.get() # 99
mem2.enable.get() # False(声明里的值还在)
class AppConfig(Config, AppKeys): # 文件版:类 + 路径
...
cfg = AppConfig("app")
ConfigMemory.__init__(data=None) 会先把类声明的默认值合进来 ——
{**type(self).DEFAULTS, **data} —— 所以 AppMemory() 拿到全部声明键与默认值,
与文件版 AppConfig(路径) 对称;data 既是初始值、也是 reset 的目标
(和文件版「default_config 交给后端」是一个道理);fname is None,_save 是空实现。
只接受一个 dict:
AppMemory(speed=60)这种关键字覆盖会当场TypeError。
唯一一条刻意的语义差别:内存版 get(缺键) 返回 None,不抛 KeyError。
主题引擎会拿内存配置去读"可能根本没有的键",读到 None 就走自己的兜底。
其余照旧 —— get('all') 返回内存快照的副本,reset(缺键) 仍然抛 KeyError。
兼容 / 快速路径:memory()
Config.memory({...}) / AppConfig.memory() 仍然可用、行为不变,但它是便捷路径:
适合「临时配置、不想为它写一个类」或者「老代码兼容」,只是替你省掉写类这一步。
mem = AppConfig.memory() # ≡ AppMemory(),默认值来自声明
mem = AppConfig.memory({"speed": 99}) # dict 盖在声明值上面,同时是 reset 目标
mem = ConfigMemory({"speed": 50}) # 连声明都没有时的裸写法
要正经当一个配置用(多键、要 reset 语义、要 IDE 补全)就定义类 ——
点操作是靠类创建时按 cls.DEFAULTS 装 property 生成的,所以"要顺手的点操作"
天然对应"有个类"。(memory() 内部照 cls.DEFAULTS 合成一个临时声明类,
所以它返回的实例也带点操作。)
动态默认值
default_config_callback 允许在初始化时和重置时动态生成默认值。
from xiaoe_config_manager import Config
def dynamic_defaults(current_config):
"""根据当前环境动态生成默认值。"""
import socket
return {
"host": "0.0.0.0",
"port": 8080,
"hostname": socket.gethostname(),
"worker_count": 4,
}
cfg = Config(
"dynamic_config",
default_config={"host": "0.0.0.0", "port": 8080,
"hostname": "", "worker_count": 4},
default_config_callback=dynamic_defaults,
init_use_callback=True, # 初始化时调用回调
)
# 重置时会再次调用回调,获取最新的默认值
cfg.reset("all")
初始化时默认值的优先级(与代码一致的说法):
default_config_callback(仅 init_use_callback=True) > default_config > 类声明的 DEFAULTS
也就是说:显式传了 default_config 时,声明里的默认值不参与
(要合并请自己合,别指望隐式规则)。
API 参考
ConfigApi
统一接口(ABC),所有配置对象都是它的实现。
| 成员 | 类型 | 说明 |
|---|---|---|
get(key) |
抽象 | 读一个键;key="all" 返回整份配置 dict。缺键抛 KeyError(内存版例外) |
set(key, value) |
抽象 | 写一个键。返回时值已生效 |
reset(key) |
抽象 | 恢复默认值;key="all" 恢复全部 |
on(key, cb) |
抽象 | 单键订阅,cb(value) |
off(key, cb) |
抽象 | 取消单键订阅(同一个函数引用)。键在、回调不在 → ValueError;键没出现过 → 静默 |
value_changed(cb) |
抽象 | 订阅所有键,cb(key, value) |
off_value_changed(cb) |
抽象 | 取消 value_changed 注册的回调(同一个函数引用)。回调不在订阅列表里 → ValueError(与 off 同一取向:注销没注册过的回调不静默) |
configget(key) |
具体 | get 的老名字,转发到 get |
configset(key, value) |
具体 | set 的老名字,转发到 set |
configreset(key) |
具体 | reset 的老名字,转发到 reset |
manager 侧
value_changed是方法(cfg.value_changed(cb),取消订阅cfg.off_value_changed(cb));xiaoe_ui的桥把同一个名字实现成 Qt 信号(cfg.value_changed.connect(slot),取消订阅.disconnect(slot))—— 桥实例上够不到off_value_changed(桥的通知走 Qt 信号,全键订阅列表永远是空的,调它必然ValueError)。
派发是逐条隔离的:
ConfigBase._dispatch对每一条回调单独try/except—— 一个回调抛异常 只打一份堆栈(_report_listener_error)、其余回调照常跑,也不会冒给set()/reset()的调用方(值那时已经落盘了)。打堆栈走的是traceback.print_exc(),sys.excepthook/threading.excepthook接不到这一份(实测:装了error_report.install_global_hooks()之后, 回调抛异常时error.log依然不多一行);console=False打包后sys.stderr is None, 这一份堆栈就等于没有 —— 想在发行版里看见,就重写ConfigBase._report_listener_error(key, callback)接到自己的错误上报里。 (xiaoe_ui的桥是另一条路:Qt 信号槽的异常由 PySide 用PyErr_Print()打出,那条路会走sys.excepthook;两边对调用方的表现仍然一致 —— 都只是打一份然后继续。)
ConfigBase
行为层:模板方法、订阅、锁、广播、唯一派发点都在这里。
Config(文件版)与 ConfigMemory(内存版)是它的平级兄弟。
| 方法 | 说明 |
|---|---|
get / set / reset |
标准三件套模板:加锁 → 子类存储原语 → 落盘/广播 → 解锁 → _dispatch() |
on / off / value_changed / off_value_changed |
订阅(_dispatch 逐条隔离:一个回调炸了不影响其余) |
_dispatch(key, value) |
唯一派发点。要接管通知(例如 Qt 桥排队回本线程),重写这一个方法 |
memory(defaults=None) |
类方法,兼容 / 快速路径:造一个内存版的自己(内部按 cls.DEFAULTS 合成临时声明类)。正经配置请定义类:class AppMem(ConfigMemory, AppKeys) |
update_config() |
把当前配置推给所有注册过的子进程队列(没队列就是空操作) |
configupdate_config(key, last_value=None) |
默认值热更新 |
item(key) |
取一个显式代理(未声明的键 / 撞保留字的键走这条) |
子类只需要实现五个存储原语:_read / _read_all / _write / _apply_reset / _save。
Config
文件版核心类,管理 JSON 文件的读写。
Config(
file_name="config",
default_config=None,
default_config_callback=None,
init_use_callback=True,
)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
file_name |
str |
"config" |
配置文件名,不含 .json |
default_config |
dict |
None |
默认配置字典,缺失的 key 自动补齐 |
default_config_callback |
Callable |
None |
默认配置回调,接受当前 config 字典作为参数,重置时再次调用 |
init_use_callback |
bool |
True |
初始化时是否用回调拿默认配置 |
方法:
| 方法 | 说明 |
|---|---|
get(key) |
获取配置值。key="all" 从文件重读并替换内存副本,返回全部配置字典;缺键抛 KeyError |
set(key, value) |
设置配置值,自动写文件并广播,然后派发通知 |
reset(key) |
恢复默认值。key="all" 恢复全部 |
configget(key) / configset(key, value) / configreset(key) |
老名字,转发到上面三个 |
configupdate_config(key, last_value=None) |
默认值热更新。用户未修改过该 key 时同步为新默认值。需预置 "{key}_reseted": False 标记,缺了抛 KeyError |
update_config() |
向所有注册的子进程广播当前配置 |
path |
property,"<file_name>.json" |
行为要点:
- 构造时只捕
FileNotFoundError;其余异常上抛,坏文件不被覆盖 - 配置项缺的键会自动补进文件(只补缺的,不覆盖已有的)
- 落盘是原子写
ConfigMemory
内存版后端。和文件版一样用定义类创建,也是对等的第二个后端(ConfigBase 的平级兄弟):
class AppMemory(ConfigMemory, AppKeys): # 类 + 声明
...
mem = AppMemory() # 用声明的默认值
mem = AppMemory({"speed": 60}) # 一份 dict 盖在声明默认值上面
| 项 | 说明 |
|---|---|
ConfigMemory(data=None) |
构造函数:{**type(self).DEFAULTS, **data} —— 声明默认值先合进来;data 既是初始值也是 reset 目标。只接受一个 dict,不支持关键字覆盖 |
fname |
None(没有文件);_save 是空实现 |
get(缺键) |
返回 None(唯一的语义差别) |
get('all') |
返回内存快照的副本 |
reset(缺键) |
抛 KeyError |
| 其余 | 与文件版一致 |
Config.memory(...) / AppConfig.memory() |
兼容 / 快速路径(行为不变):临时配置不想写类时用它。有声明键时返回 ConfigMemory 的一个合成子类(点操作照带) |
ConfigKeysBase / ConfigDeclareMixin / ConfigItemProxy[T]
| 名字 | 说明 |
|---|---|
ConfigKeysBase |
纯声明基类,DEFAULTS 是普通 dict,可以被任何后端消费 |
ConfigDeclareMixin |
把声明键变成实例上的 property。ConfigBase 已继承 |
ConfigItemProxy[T] |
点操作代理,提供 .get() / .set(v) / .reset() / .on(cb) / .off(cb) 和 .key |
collect_defaults(cls) |
扫出一个类认哪些键、默认值各是多少 |
default(value) / default(None, type_) |
声明默认值的辅助函数,也承担类型推断 |
ConfigItemProxy 由 ConfigDeclareMixin 通过类属性自动生成,
不能在代码里手动实例化。
default
default(30) # 推断为 ConfigItemProxy[int]
default("myapp") # 推断为 ConfigItemProxy[str]
default(None, list) # 推断为 ConfigItemProxy[list | None]
default(None, dict) # 推断为 ConfigItemProxy[dict | None]
运行时行为:直接返回传入的 value(type_ 参数仅用于类型推断)。
ConfigInMainProcessPipe
主进程管道接口,每个子进程对应创建一个实例。
ConfigInMainProcessPipe(config) # config: 主进程里的配置单例
| 方法 | 说明 |
|---|---|
get_queues() |
返回 (set_queue, get_queue) 元组,传递给子进程 |
close() |
关闭管道,停止监听线程 |
它把队列注册进 config.queues(于是主进程的每次 set 都会自动广播),
并起一个线程收子进程的消息;收到后直接调 config.set / config.reset,
不另外发通知(否则就是双发)。
ConfigInChildProcessPipe
子进程侧的配置实例,通过队列与主进程通信。与 Config 是同一套接口
(点操作、on / off / value_changed / off_value_changed 都在),继承 ConfigDeclareMixin,
配一份声明就有 cfg.xxx.get()。
ConfigInChildProcessPipe(config_set_queue, config_get_queue)
| 方法 | 说明 |
|---|---|
get(key) |
从本地缓存读取;缺键抛 KeyError |
get("all") |
返回 None —— 本地缓存里没有 "all" 这个键。要枚举键请用声明 |
set(key, value) |
修改配置(乐观:先改本地缓存,消息发给主进程落盘+广播) |
reset(key) |
恢复默认值(悲观/异步:只发消息,本地缓存要等广播回来才变) |
on / off / value_changed / off_value_changed |
订阅,跨进程的变更由 diff 自动派发到这里 |
close() |
关闭管道,停止同步线程 |
启动:构造时会向主进程要一份完整配置并等它回来,
等 INIT_TIMEOUT = 30 秒还没到就抛 RuntimeError —— 说明主进程的
ConfigInMainProcessPipe 还没起来,或者管道对象传错了。
reset的异步性是设计,不是 bug:跨进程的写一律经主进程统一处理。 代价是reset返回时读到的可能还是旧值,广播到达时会按 diff 派发正确的新值。
diff_changes(before, after)
[(key, value), ...],逐键 diff。用它才能回答"整份配置广播下来,到底哪些键变了"。
注意事项
- 多进程环境下请保持
Config单例 — 多个Config实例同时操作同一文件会导致数据不一致 - 子进程的
ConfigInChildProcessPipe不直接写文件 — 所有写操作会发送请求,由主进程统一处理 - 多进程场景推荐使用共享声明类模式 — 把配置声明抽到
AppKeys,主进程与子进程各自继承,避免两边维护两份键名 - 单进程场景直接继承
Config+ 声明类 — 无需为多进程专门拆类 - 使用完毕后调用
close()— 主进程的ConfigInMainProcessPipe与子进程的ConfigInChildProcessPipe都要调,避免守护线程未正常退出 - 文件名不含
.json后缀 — 传入"myconfig"对应myconfig.json - 父目录必须先存在 —
Config只负责写"<file_name>.json",不建目录。"config_files/app"这种带层级的路径要先自己os.makedirs(xiaoe_ui的app_config_dir()会代劳),否则补默认值 /set()落盘时抛FileNotFoundError - 子进程的
get('all')返回None— 需要整份配置时请在主进程侧读,或在子进程里用声明逐个键读 - 老名字不要拿去覆写 — 子类实现
get/set/reset;configget/configset/configreset是接口层的转发壳 get('all')有重读文件并替换内存副本的副作用 — 要读单键就用get(key)
许可证
MIT
Metadata
Release files for xiaoe-config-manager 2.1.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| xiaoe_config_manager-2.1.1.tar.gz | 52.0 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| xiaoe_config_manager-2.1.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 86.9 kB
Release files / xiaoe_config_manager-2.1.1.tar.gz
| Download URL | xiaoe_config_manager-2.1.1.tar.gz |
|---|---|
| Size | 52.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
9904b8827be33ac62d788164803bc79c66046fb8188b1046ecda1debc391b80a
|
|
BLAKE2b-256 checksum How to use checksums |
7e7b966bd14a7ccc0fcc5413069661bad08d6fe57a5c1eedaec6bcae5df2fd5c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.9
|
Release files / xiaoe_config_manager-2.1.1-py3-none-any.whl
| Download URL | xiaoe_config_manager-2.1.1-py3-none-any.whl |
|---|---|
| Size | 35.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
9c234b2ffa9ec7c0b1105d20195f32177a2f314977b41fa4be0072697dcc8995
|
|
BLAKE2b-256 checksum How to use checksums |
7dba7efcc87fc4add9ffc7b29e01702b1991cfd6fd449eb79f43606ab872bf5b
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.9
|
Security for xiaoe-config-manager 2.1.1
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.