from __future__ import annotations
import datetime
import hashlib
import json
import logging
import re
import threading
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
@dataclass(frozen=True)
class Rule:
"""
单条评估规则。
- name: 规则名称，便于调试与导出
- predicate: 上下文判断函数，入参为 context: Dict[str, Any]，返回 True/False
- rollout: 可选的按比例灰度（0-100），当 predicate 为 True 且 identity 命中 rollout 桶时才算匹配
"""
name: str
predicate: Callable[[Dict[str, Any]], bool]
rollout: Optional[int] = None
def matches(self, context: Optional[Dict[str, Any]], key: str) -> bool:
"""
计算规则对给定上下文是否命中。
- 若设置 rollout，需要上下文中可解析的 identity；若缺失 identity，则不命中。
"""
ctx = context or {}
try:
if not self.predicate(ctx):
return False
except Exception:
# 避免业务 predicate 抛错影响整体评估
return False
if self.rollout is None:
return True
identity = FeatureToggleManager._extract_identity(ctx)
if identity is None:
return False # 没有 identity 时，带 rollout 的规则不命中，避免意外放量
bucket = FeatureToggleManager.compute_rollout_bucket(key, identity)
return bucket < int(self.rollout)
@dataclass
class Toggle:
"""
单个功能开关的定义。
- key: 唯一键
- enabled: 顶层开关（全局门禁）。False 时无论规则如何均返回 False；True 时结合规则评估。
- rules: 可选规则列表；存在时需至少一条规则命中才返回 True。
"""
key: str
enabled: bool
rules: Optional[List[Rule]] = None
class FeatureToggleManager:
"""
统一管理功能开关的定义、加载、刷新与评估，支持：
- 本地配置与远程拉取（fetcher）
- 用户分群与灰度（Rule.predicate + Rule.rollout）
- 可插拔钩子通知（开关 enabled 基态变化）
- 线程安全缓存（TTL + RLock）
适用场景：按环境发布、A/B 实验预埋、回滚策略等。
"""
# -------------------------
# 公共属性（按要求暴露）
# -------------------------
toggles: Dict[str, Toggle]
default_state: bool
fetcher: Optional[Callable[[], Dict[str, Any]]]
cache_ttl_seconds: int
last_refresh_at: Optional[datetime.datetime]
hooks: List[Callable[[str, bool], None]]
lock: threading.RLock
logger: logging.Logger
# -------------------------
# 初始化
# -------------------------
def __init__(
self,
default_state: bool = False,
fetcher: Optional[Callable[[], Dict[str, Any]]] = None,
cache_ttl_seconds: int = 30,
logger: Optional[logging.Logger] = None,
initial_toggles: Optional[Dict[str, Toggle]] = None,
) -> None:
"""
参数:
- default_state: bool，未命中时的默认值
- fetcher: 可选远程拉取函数，返回 Dict[str, Any] 的最新开关定义
- cache_ttl_seconds: 本地缓存 TTL，过期后触发刷新
- logger: 日志记录器，默认使用模块 logger
- initial_toggles: 可选初始开关集合
"""
self.default_state = bool(default_state)
self.fetcher = fetcher
self.cache_ttl_seconds = int(cache_ttl_seconds)
self.logger = logger or logging.getLogger(__name__)
self.toggles = initial_toggles.copy() if initial_toggles else {}
self.last_refresh_at = None
self.hooks = []
self.lock = threading.RLock()
# -------------------------
# 核心 API
# -------------------------
def is_enabled(
self,
key: str,
context: Optional[Dict[str, Any]] = None,
default: Optional[bool] = None,
) -> bool:
"""
基于规则与上下文计算开关是否开启（线程安全）。
评估规则：
- 若 key 不存在 => 返回 fallback: default 或 default_state
- 若 toggle.enabled 为 False => 返回 False
- 若 toggle.enabled 为 True:
- 无规则 => True
- 存在规则 => 任一规则命中且（若配置 rollout）命中桶 => True；否则 False
参数:
- key: str，开关键
- context: 可选 Dict，上下文（如 user_id、country、app_version 等）
- default: 可选 bool，当 key 不存在时的回退值（优先于 default_state）
返回:
- bool，是否开启
"""
# 自动确保在 TTL 过期时刷新缓存
self._ensure_fresh()
with self.lock:
toggle = self.toggles.get(key)
if toggle is None:
return bool(default if default is not None else self.default_state)
if not toggle.enabled:
return False
if not toggle.rules:
return True
for rule in toggle.rules:
try:
if rule.matches(context, key):
return True
except Exception as e:
self.logger.exception("Rule evaluation failed for %s/%s: %s", key, rule.name, e)
return False
def set_toggle(
self,
key: str,
state: bool,
rules: Optional[List[Rule]] = None,
) -> None:
"""
设置或更新单个开关，触发变更钩子（仅当 enabled 基态变化时）。
参数:
- key: str
- state: bool，enabled 基态
- rules: 可选规则列表
返回:
- None
"""
with self.lock:
old = self.toggles.get(key)
old_enabled = old.enabled if old else None
self.toggles[key] = Toggle(key=key, enabled=bool(state), rules=rules[:] if rules else None)
hooks_to_call = []
if old_enabled is None or bool(state) != bool(old_enabled):
hooks_to_call = self.hooks[:]
# 钩子在锁外调用，避免阻塞/死锁
for cb in hooks_to_call:
try:
cb(key, bool(state))
except Exception as e:
self.logger.exception("Hook callback failed for %s: %s", key, e)
def refresh(self, force: bool = False) -> bool:
"""
在 TTL 过期或强制模式下调用 fetcher 拉取最新开关，成功返回 True。
参数:
- force: bool，是否强制刷新（忽略 TTL）
返回:
- bool，是否发生了刷新且拉取成功
"""
if self.fetcher is None:
return False
now = datetime.datetime.utcnow()
with self.lock:
if not force and self.last_refresh_at is not None:
age = (now - self.last_refresh_at).total_seconds()
if age < self.cache_ttl_seconds:
return False # 未过期，无需刷新
# 调用远程拉取在锁外执行，避免长时间占用锁
try:
raw = self.fetcher() or {}
except Exception as e:
self.logger.exception("Fetcher failed: %s", e)
return False
# 解析并计算变化
new_toggles, changed_keys = self._parse_remote_payload(raw)
with self.lock:
# 对比 enabled 基态的变化（用于钩子）
enabled_changes = self._diff_enabled_changes(self.toggles, new_toggles)
self.toggles = new_toggles
self.last_refresh_at = now
hooks_to_call = self.hooks[:]
changed_keys = enabled_changes # 仅对 enabled 基态变化触发钩子
for key, enabled in changed_keys:
for cb in hooks_to_call:
try:
cb(key, enabled)
except Exception as e:
self.logger.exception("Hook callback failed for %s: %s", key, e)
return True
def load_from_file(self, path: str) -> int:
"""
从本地 JSON/YAML 文件加载开关定义，返回加载数量。
文件格式支持两种顶层结构：
1) 直接 key -> toggleDef
2) {"toggles": { key -> toggleDef }}
toggleDef 支持：
- 布尔值：true/false
- 对象：
{
"enabled": true/false,
"rules": [
{
"name": "rule-name",
"predicate": "always" | { ... 见 _compile_predicate 文档 ... },
"rollout": 30
}
]
}
参数:
- path: str，文件路径（.json/.yaml/.yml）
返回:
- int，成功加载的开关数量
"""
content: Dict[str, Any]
if path.lower().endswith(".json"):
with open(path, "r", encoding="utf-8") as f:
content = json.load(f)
elif path.lower().endswith((".yml", ".yaml")):
try:
import yaml # type: ignore
except Exception as e:
raise RuntimeError("PyYAML is required to load YAML files") from e
with open(path, "r", encoding="utf-8") as f:
content = yaml.safe_load(f) or {}
else:
raise ValueError("Unsupported file extension. Use .json, .yaml or .yml")
toggles_obj = content.get("toggles", content)
if not isinstance(toggles_obj, dict):
raise ValueError("Invalid toggle file format")
new_toggles: Dict[str, Toggle] = {}
for key, val in toggles_obj.items():
t = self._parse_toggle_entry(str(key), val)
new_toggles[key] = t
with self.lock:
enabled_changes = self._diff_enabled_changes(self.toggles, new_toggles)
self.toggles.update(new_toggles)
hooks_to_call = self.hooks[:]
for key, enabled in enabled_changes:
for cb in hooks_to_call:
try:
cb(key, enabled)
except Exception as e:
self.logger.exception("Hook callback failed for %s: %s", key, e)
return len(new_toggles)
def register_hook(self, cb: Callable[[str, bool], None]) -> None:
"""
注册开关 enabled 基态变化的通知回调。
参数:
- cb: Callable[[str, bool], None]，签名为 (key, enabled)
返回:
- None
"""
if not callable(cb):
raise ValueError("hook must be callable")
with self.lock:
self.hooks.append(cb)
def export_state(self) -> Dict[str, Any]:
"""
导出当前开关快照，便于调试或持久化（不导出 predicate 可执行体，仅导出规则元信息）。
返回:
- Dict[str, Any]，包含 toggles、default_state、cache_ttl_seconds、last_refresh_at
"""
with self.lock:
snapshot: Dict[str, Any] = {
"default_state": self.default_state,
"cache_ttl_seconds": self.cache_ttl_seconds,
"last_refresh_at": self.last_refresh_at.isoformat() if self.last_refresh_at else None,
"toggles": {},
}
for k, t in self.toggles.items():
snapshot["toggles"][k] = {
"enabled": t.enabled,
"rules": [{"name": r.name, "rollout": r.rollout} for r in (t.rules or [])],
}
return snapshot
# -------------------------
# 线程安全缓存辅助
# -------------------------
def _ensure_fresh(self) -> None:
"""
若 TTL 过期尝试刷新（非强制），fetcher 可能为 None。
"""
if self.fetcher is None or self.cache_ttl_seconds <= 0:
return
need_refresh = False
now = datetime.datetime.utcnow()
with self.lock:
if self.last_refresh_at is None:
need_refresh = True
else:
age = (now - self.last_refresh_at).total_seconds()
if age >= self.cache_ttl_seconds:
need_refresh = True
if need_refresh:
try:
self.refresh(force=False)
except Exception as e:
self.logger.exception("Auto refresh failed: %s", e)
# -------------------------
# 解析与编译
# -------------------------
def _parse_remote_payload(self, raw: Dict[str, Any]) -> Tuple[Dict[str, Toggle], List[Tuple[str, bool]]]:
"""
将 fetcher 返回的原始字典解析为 Toggle 集合。
返回: (new_toggles, changed_enabled_pairs)，其中 changed_enabled_pairs 在此阶段不计算，
实际“变化”由 _diff_enabled_changes 决定；此处返回空列表即可。
"""
toggles_obj = raw.get("toggles", raw)
new_toggles: Dict[str, Toggle] = {}
if not isinstance(toggles_obj, dict):
self.logger.warning("Fetcher payload is not a dict")
return {}, []
for key, val in toggles_obj.items():
try:
t = self._parse_toggle_entry(str(key), val)
new_toggles[key] = t
except Exception as e:
self.logger.exception("Failed to parse toggle %s: %s", key, e)
return new_toggles, []
def _parse_toggle_entry(self, key: str, val: Any) -> Toggle:
"""
解析单个 toggle 条目。
支持：
- bool => enabled=val, rules=None
- dict => 需要 enabled 字段，可含 rules
"""
if isinstance(val, bool):
return Toggle(key=key, enabled=val, rules=None)
if isinstance(val, dict):
if "enabled" not in val:
raise ValueError(f"Toggle {key} missing 'enabled'")
enabled = bool(val.get("enabled"))
raw_rules = val.get("rules") or []
rules: List[Rule] = []
for r in raw_rules:
if not isinstance(r, dict):
raise ValueError(f"Toggle {key} rules item must be object")
name = str(r.get("name") or "rule")
rollout = r.get("rollout")
if rollout is not None:
rollout = int(rollout)
if rollout < 0 or rollout > 100:
raise ValueError(f"Toggle {key} rule {name} rollout must be 0..100")
pred_spec = r.get("predicate", "always")
predicate = self._compile_predicate(pred_spec)
rules.append(Rule(name=name, predicate=predicate, rollout=rollout))
return Toggle(key=key, enabled=enabled, rules=rules or None)
raise ValueError(f"Unsupported toggle definition for {key}")
# 简易谓词编译器：支持常见条件组合，满足多数灰度/分群场景
def _compile_predicate(self, spec: Any) -> Callable[[Dict[str, Any]], bool]:
"""
支持的 spec 形式：
- "always" => 恒真
- {"eq": {"field": "country", "value": "US"}}
- {"in": {"field": "country", "values": ["US", "CA"]}}
- {"regex": {"field": "app_version", "pattern": "^2\\."}}
- {"gte": {"field": "build", "value": 120}} 数字比较
- {"any": [spec1, spec2, ...]}
- {"all": [spec1, spec2, ...]}
- {"not": spec}
"""
if spec == "always" or spec is True or spec is None:
return lambda ctx: True
if not isinstance(spec, dict) or len(spec) != 1:
raise ValueError(f"Unsupported predicate spec: {spec}")
op, arg = next(iter(spec.items()))
if op == "eq":
field, value = arg["field"], arg["value"]
return lambda ctx: ctx.get(field) == value
if op == "in":
field, values = arg["field"], set(arg["values"])
return lambda ctx: ctx.get(field) in values
if op == "regex":
field, pattern = arg["field"], arg["pattern"]
regex = re.compile(pattern)
return lambda ctx: isinstance(ctx.get(field), str) and bool(regex.search(ctx[field]))
if op == "gte":
field, value = arg["field"], arg["value"]
return lambda ctx: FeatureToggleManager._to_number(ctx.get(field)) >= FeatureToggleManager._to_number(value)
if op == "any":
subs = [self._compile_predicate(s) for s in (arg or [])]
return lambda ctx: any(p(ctx) for p in subs)
if op == "all":
subs = [self._compile_predicate(s) for s in (arg or [])]
return lambda ctx: all(p(ctx) for p in subs)
if op == "not":
sub = self._compile_predicate(arg)
return lambda ctx: not sub(ctx)
raise ValueError(f"Unknown predicate operator: {op}")
# -------------------------
# 工具方法
# -------------------------
@staticmethod
def _to_number(v: Any) -> float:
try:
return float(v)
except Exception:
return float("nan")
@staticmethod
def _extract_identity(context: Dict[str, Any]) -> Optional[str]:
"""
从上下文中提取稳定 identity，用于一致性灰度。
常用字段：user_id/uid/id/device_id/session_id/account/email
"""
candidates = ("user_id", "uid", "id", "device_id", "session_id", "account", "email")
for k in candidates:
if k in context and context[k] is not None:
return str(context[k])
return None
@staticmethod
def compute_rollout_bucket(key: str, identity: str, buckets: int = 100) -> int:
"""
基于 key + identity 生成 0..(buckets-1) 的稳定桶值（默认 0..99）。
"""
seed = f"{key}:{identity}".encode("utf-8")
digest = hashlib.sha256(seed).hexdigest()
as_int = int(digest[:8], 16) # 取前 32bit 足够
return as_int % buckets
@staticmethod
def _diff_enabled_changes(
old: Dict[str, Toggle],
new: Dict[str, Toggle],
) -> List[Tuple[str, bool]]:
"""
计算 enabled 基态的变化，用于触发钩子。
返回: [(key, new_enabled), ...]
"""
changes: List[Tuple[str, bool]] = []
old_keys = set(old.keys())
new_keys = set(new.keys())
for k in new_keys.union(old_keys):
old_enabled = old.get(k).enabled if k in old else None
new_enabled = new.get(k).enabled if k in new else None
if old_enabled is None and new_enabled is not None:
changes.append((k, bool(new_enabled)))
elif new_enabled is None:
# 新集合中不存在，认为删除，不触发（也可选择触发 False）
continue
elif bool(old_enabled) != bool(new_enabled):
changes.append((k, bool(new_enabled)))
return changes
