consul-config

基于 Consul KV 的动态配置管理库（Leader/Follower + 本地文件广播）。

发行名 consul-config · import consul_config（与 consul-http 对称命名）

适用 / 不适用

本包面向「Consul KV + 单机多 worker（或每机独立选主）」场景，不是通用配置中心银弹。

适用

• 配置源为 Consul KV（或可由运维写入 KV）
• 部署为 单机多 worker（如 uvicorn --workers N）：同机进程共享本地缓存文件，仅 Leader watch Consul
• 或多机部署但 每台机各自选主、各自本地缓存（机与机之间不共享内存配置）
• 配置项可经 Schema / Registry 校验；非法值保留上次成功值，Consul 不可达时回退调用方默认值（如 .env）
• Token 具备：prefix 读、锁 key 写、session:write
• Python 3.10+，Consul 1.10+ HTTP API

不适用

• 多机共享同一份热更新配置且不能（或不想）每台机独立打 Consul / 写本地盘 —— 需 Redis 等共享存储或其它配置中心
• 需要 跨机强一致、配置变更的分布式事务 / CAS 编排（本包不做 Txn）
• 服务发现、健康检查、Connect、Agent/Catalog 等 —— 用完整 Consul SDK；HTTP KV/Session 客户端见 consul-http
• 要求 零 ACL、零本地磁盘、零选主 的纯只读轮询方案（那是另一类设计）

Leader 故障后接管时间上界约为 leader_ttl_seconds + follower_poll_seconds。

特性

• 优先级：Consul KV > 调用方默认值（如 .env）
• 实时更新：Leader 用 blocking query 监听前缀，秒级感知变更
• 多 worker 只更新一次：Leader 经 Consul 锁选举，校验后原子广播到本地缓存；Follower 只轮询文件
• 接入 DX：Settings mixin、ManagerConfig.from_settings、ConfigResolver、ConfigRegistry、FastAPI setup_dynamic_config、consul-config probe

安装

pip install consul-config
# FastAPI 集成（可选）
pip install "consul-config[fastapi]"
# 或
uv add consul-config

依赖：consul-http>=0.1.0,<0.2、pydantic>=2。

Monorepo 可编辑依赖示例：

[project]
dependencies = ["consul-http", "consul-config"]

[tool.uv.sources]
consul-http = { path = "packages/consul_http", editable = true }
consul-config = { path = "packages/consul_config", editable = true }

快速接入（推荐）

from pydantic_settings import BaseSettings
from consul_config import (
    ConfigManager,
    ConfigRegistry,
    ConfigResolver,
    DynamicConfigSettingsMixin,
    ManagerConfig,
)

class Settings(DynamicConfigSettingsMixin, BaseSettings):
    # 只需应用名：自动推导 app/myapp/ 与 app/myapp-config/leader
    dynamic_config_app_name: str = "myapp"
    consul_http_addr: str = "http://127.0.0.1:8500"
    consul_enabled: bool = True

settings = Settings()

registry = ConfigRegistry()
registry.bool("dark_mode", kv="flags/dark_mode", default=False)
registry.int("rate_per_minute", kv="limits/rate", default=30, ge=0)

manager = ConfigManager(
    ManagerConfig.from_settings(settings),
    registry.build_schema(),
)
manager.start()

resolver = ConfigResolver(manager)
dark = resolver.get("dark_mode", default=False)

FastAPI 一行挂载

from fastapi import FastAPI
from consul_config.integrations.fastapi import setup_dynamic_config, get_manager

app = FastAPI()
setup_dynamic_config(app, schema=registry.build_schema(), settings=settings)
# lifespan 内自动 start/stop；请求内：get_manager(app).get("dark_mode")

CLI 探针（验证 ACL / 选主）

# 读取 .env 中的 CONSUL_*（不打印 token）
uv run consul-config probe
# 或：uv run python -m consul_config probe

KV 布局约定

app/myapp/            <- prefix（Leader 监听）
  flags/dark_mode
app/myapp-config/     <- prefix 外
  leader              <- 分布式锁

运行参数

本地开发

uv sync
uv run pytest
uv run pytest -m integration   # 需要 Consul
uv run ruff check .