—— 配置文件的黄金搭档
import json
# 美化输出
data = {"name": "张三", "skills": ["Python", "Go"]}
pretty = json.dumps(data, ensure_ascii=False, indent=2)
print(pretty)
# 自定义序列化
from datetime import datetime, date
from decimal import Decimal
from pathlib import Path
class SmartEncoder(json.JSONEncoder):
"""智能 JSON 编码器"""
def default(self, obj):
if isinstance(obj, (datetime, date)):
return obj.isoformat()
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, Path):
return str(obj)
if isinstance(obj, set):
return sorted(list(obj))
if isinstance(obj, bytes):
return obj.decode("utf-8")
return super().default(obj)
# 使用
data = {
"created": datetime.now(),
"price": Decimal("99.99"),
"tags": {"python", "go"},
"path": Path("/tmp/test"),
}
encoded = json.dumps(data, cls=SmartEncoder, ensure_ascii=False, indent=2)
# 自定义反序列化
def smart_decode(d):
"""自动转换 ISO 日期字符串"""
for k, v in d.items():
if isinstance(v, str):
try:
d[k] = datetime.fromisoformat(v)
except ValueError:
pass
return dimport yaml # pip install pyyaml
# YAML 字符串 → Python
yaml_str = """
database:
host: localhost
port: 3306
name: mydb
credentials:
username: admin
password: secret
logging:
level: INFO
handlers:
- console
- file
features:
cache: true
retry: 3
"""
config = yaml.safe_load(yaml_str)
print(config["database"]["host"]) # localhost
print(config["logging"]["handlers"]) # ['console', 'file']
# Python → YAML 字符串
data = {
"server": {
"host": "0.0.0.0",
"port": 8080,
"debug": False,
},
"routes": [
{"path": "/api", "method": "GET"},
{"path": "/submit", "method": "POST"},
],
}
yaml_output = yaml.dump(data, allow_unicode=True, default_flow_style=False)
print(yaml_output)
# 读写 YAML 文件
with open("config.yaml", "w", encoding="utf-8") as f:
yaml.dump(data, f, allow_unicode=True, default_flow_style=False)
with open("config.yaml", "r", encoding="utf-8") as f:
loaded = yaml.safe_load(f)import json
import yaml
from pathlib import Path
from typing import Any
class ConfigManager:
"""多环境配置管理器"""
def __init__(self, config_dir: str = "config"):
self.config_dir = Path(config_dir)
self._config = {}
def load(self, env: str = "dev"):
"""
加载配置: base.yaml + {env}.yaml → 合并
环境配置覆盖基础配置
"""
base_file = self.config_dir / "base.yaml"
env_file = self.config_dir / f"{env}.yaml"
self._config = {}
if base_file.exists():
with open(base_file, encoding="utf-8") as f:
self._config = yaml.safe_load(f) or {}
if env_file.exists():
with open(env_file, encoding="utf-8") as f:
env_config = yaml.safe_load(f) or {}
self._deep_merge(self._config, env_config)
return self
def get(self, key: str, default: Any = None) -> Any:
"""支持点号分隔的键: 'database.host'"""
keys = key.split(".")
value = self._config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
else:
return default
if value is None:
return default
return value
@staticmethod
def _deep_merge(base: dict, override: dict) -> dict:
"""深度合并字典"""
for k, v in override.items():
if k in base and isinstance(base[k], dict) and isinstance(v, dict):
ConfigManager._deep_merge(base[k], v)
else:
base[k] = v
return base
# 使用
# cfg = ConfigManager("config").load("production")
# db_host = cfg.get("database.host", "localhost")
# db_port = cfg.get("database.port", 3306)import json
def validate_config(data: dict, required_keys: list[str]) -> list[str]:
"""简单配置验证（不依赖第三方库）"""
errors = []
for key_path in required_keys:
keys = key_path.split(".")
current = data
for key in keys:
if isinstance(current, dict) and key in current:
current = current[key]
else:
errors.append(f"缺少必需配置: {key_path}")
break
return errors
# 使用
required = ["database.host", "database.port", "logging.level"]
config = {"database": {"host": "localhost", "port": 3306}}
errors = validate_config(config, required)
# ["缺少必需配置: logging.level"]#!/usr/bin/env python3
"""第11课 JSON/YAML验证"""
import json
import tempfile
import os
from datetime import datetime
from decimal import Decimal
from pathlib import Path
def test_json_advanced():
"""JSON高级操作测试"""
class SmartEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, Path):
return str(obj)
if isinstance(obj, set):
return sorted(list(obj))
return super().default(obj)
data = {"ts": datetime(2026,1,1), "price": Decimal("9.99"),
"tags": {"a","b"}, "path": Path("/tmp")}
encoded = json.dumps(data, cls=SmartEncoder)
parsed = json.loads(encoded)
assert parsed["price"] == 9.99
assert parsed["tags"] == ["a", "b"]
print("✅ JSON高级操作测试通过")
def test_json_file():
"""JSON文件读写测试"""
data = {"name": "张三", "scores": [95,88]}
with tempfile.NamedTemporaryFile(suffix=".json", delete=False,
mode="w", encoding="utf-8") as f:
path = f.name
json.dump(data, f, ensure_ascii=False)
with open(path, "r", encoding="utf-8") as f:
loaded = json.load(f)
assert loaded["name"] == "张三"
os.unlink(path)
print("✅ JSON文件读写测试通过")
def test_yaml():
"""YAML读写测试"""
import yaml
yaml_str = "host: localhost\nport: 3306\n"
config = yaml.safe_load(yaml_str)
assert config["host"] == "localhost"
assert config["port"] == 3306
output = yaml.dump(config, allow_unicode=True)
parsed = yaml.safe_load(output)
assert parsed["host"] == "localhost"
print("✅ YAML读写测试通过")
def test_config_merge():
"""配置合并测试"""
def deep_merge(base, override):
for k, v in override.items():
if k in base and isinstance(base[k], dict) and isinstance(v, dict):
deep_merge(base[k], v)
else:
base[k] = v
return base
base = {"db": {"host": "localhost", "port": 3306}, "debug": True}
override = {"db": {"host": "prod-db"}, "debug": False}
result = deep_merge(base, override)
assert result["db"]["host"] == "prod-db"
assert result["db"]["port"] == 3306
assert result["debug"] == False
print("✅ 配置合并测试通过")
if __name__ == "__main__":
test_json_advanced()
test_json_file()
test_yaml()
test_config_merge()
print("\n🎉 第11课全部验证通过！")