文章
8 · 模块化构建：标准库与自定义包
阅读数据加载中…
点赞数据加载中…
🔗 知识图谱导航：阅读本文前，建议先掌握/回顾 《7 · 面向对象：封装、继承与系统解耦》 中的核心概念；本文会在这个基础上继续推进。 承上启下：前7篇我们写的都是单文件脚本。真实项目需要多文件协作——模块化是从"能跑"到"可维护"的关键一跃。
极客解析：先把数据流、控制流和模块边界跑通，再谈抽象；每段代码都围绕一个可执行 CLI 闭环展开。
痛点与架构
geek_toolkit/ ← 自定义包（目录）
__init__.py ← 包的入口，控制对外暴露的接口
config.py ← 配置管理（os.environ + dataclass）
logger.py ← 日志工具（logging 标准库）
timer.py ← 计时工具（time + functools）
cli.py ← CLI 入口（argparse）
main.py ← 调用包的主程序
| 标准库模块 | 用途 | 替代第三方库 |
|---|---|---|
pathlib |
路径操作 | os.path |
logging |
结构化日志 | loguru（可选） |
argparse |
CLI 参数 | click（可选） |
dataclasses |
数据类 | pydantic（可选） |
json |
JSON 序列化 | 无需替代 |
tempfile |
临时文件 | 无需替代 |
实战演练场
完整的自包含演示（单文件模拟多模块结构）：
步步为营：核心逻辑自适应拆解
模块化不是把代码拆散，而是把职责分清。下面按“配置 -> 日志 -> 路径 -> JSON -> CLI 调度”的顺序拆开，你会看到每个标准库模块负责哪一块，最后再由 main() 把它们组装成完整工具。
Step 1：用 AppConfig 把散落配置收进一个对象
痛点与机制：
AppConfig 解决的是“配置到处散落”的问题。环境变量像外部遥控器，默认值像备用按钮：有环境变量就优先用，没有就回到代码里的默认配置。to_dict() 则把 Path 这种对象转成更容易展示和序列化的字符串。
核心源码（逐字来自文末完整源码）：
@dataclass
class AppConfig:
"""应用配置：优先读取环境变量，其次使用默认值。"""
app_name: str = field(default_factory=lambda: os.environ.get("APP_NAME", "geek-toolkit"))
log_level: str = field(default_factory=lambda: os.environ.get("LOG_LEVEL", "INFO"))
data_dir: Path = field(default_factory=lambda: Path(os.environ.get("DATA_DIR", "./data")))
max_workers: int = field(default_factory=lambda: int(os.environ.get("MAX_WORKERS", "4")))
debug: bool = field(default_factory=lambda: os.environ.get("DEBUG", "").lower() == "true")
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["data_dir"] = str(self.data_dir)
return d
可运行演示（补齐 Mock 数据与 print 反馈）：
import os
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
@dataclass
class AppConfig:
"""应用配置：优先读取环境变量，其次使用默认值。"""
app_name: str = field(default_factory=lambda: os.environ.get("APP_NAME", "geek-toolkit"))
log_level: str = field(default_factory=lambda: os.environ.get("LOG_LEVEL", "INFO"))
data_dir: Path = field(default_factory=lambda: Path(os.environ.get("DATA_DIR", "./data")))
max_workers: int = field(default_factory=lambda: int(os.environ.get("MAX_WORKERS", "4")))
debug: bool = field(default_factory=lambda: os.environ.get("DEBUG", "").lower() == "true")
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["data_dir"] = str(self.data_dir)
return d
cfg = AppConfig()
print("默认配置:", cfg.to_dict())
os.environ["APP_NAME"] = "demo-app"
os.environ["MAX_WORKERS"] = "8"
print("环境变量覆盖后:", AppConfig().to_dict())
del os.environ["APP_NAME"]
del os.environ["MAX_WORKERS"]
Step 2：用 demo_config 把配置优先级打印给人看
痛点与机制：
光有配置类还不够，新手需要看到它到底读出了什么。demo_config() 用表格把每个配置项打印出来，并临时设置 APP_NAME 演示“环境变量优先”的效果。运行完再删除环境变量，避免影响后面的代码。
核心源码（逐字来自文末完整源码）：
def demo_config() -> None:
cfg = AppConfig()
print("\n ── 配置管理演示 ──────────────────────────")
print(f" {'配置项':<15} {'值'}")
print(f" {'─'*15} {'─'*25}")
for k, v in cfg.to_dict().items():
print(f" {k:<15} {v}")
# 环境变量覆盖演示
os.environ["APP_NAME"] = "my-custom-app"
cfg2 = AppConfig()
print(f"\n 设置 APP_NAME 环境变量后: {cfg2.app_name}")
del os.environ["APP_NAME"]
可运行演示（补齐 Mock 数据与 print 反馈）：
import os
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
@dataclass
class AppConfig:
"""应用配置：优先读取环境变量，其次使用默认值。"""
app_name: str = field(default_factory=lambda: os.environ.get("APP_NAME", "geek-toolkit"))
log_level: str = field(default_factory=lambda: os.environ.get("LOG_LEVEL", "INFO"))
data_dir: Path = field(default_factory=lambda: Path(os.environ.get("DATA_DIR", "./data")))
max_workers: int = field(default_factory=lambda: int(os.environ.get("MAX_WORKERS", "4")))
debug: bool = field(default_factory=lambda: os.environ.get("DEBUG", "").lower() == "true")
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["data_dir"] = str(self.data_dir)
return d
# Step 2：演示函数像说明书，把配置项和值排成表格。
def demo_config() -> None:
cfg = AppConfig()
print("\n ── 配置管理演示 ──────────────────────────")
print(f" {'配置项':<15} {'值'}")
print(f" {'─'*15} {'─'*25}")
for k, v in cfg.to_dict().items():
print(f" {k:<15} {v}")
# 环境变量覆盖演示
os.environ["APP_NAME"] = "my-custom-app"
cfg2 = AppConfig()
print(f"\n 设置 APP_NAME 环境变量后: {cfg2.app_name}")
del os.environ["APP_NAME"]
demo_config()
Step 3：用 setup_logger 创建可复用的结构化日志器
痛点与机制：
print() 适合临时观察，logging 适合项目长期运行。setup_logger() 统一设置日志级别、输出位置和时间格式，就像给团队规定统一日报模板，后面哪个模块用它，输出都整齐。
核心源码（逐字来自文末完整源码）：
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""配置结构化日志器。"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
fmt=" %(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
可运行演示（补齐 Mock 数据与 print 反馈）：
import logging
import sys
# Step 3：日志器像统一广播站，所有模块都按同一种格式喊话。
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""配置结构化日志器。"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
fmt=" %(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
logger = setup_logger("demo.pipeline", level="DEBUG")
logger.debug("调试信息：准备读取配置")
logger.info("任务开始：模块化演示")
logger.warning("这是一个警告示例")
Step 4：用 demo_logging 跑一组真实日志级别
痛点与机制：
debug 像开发时的草稿备注，info 是正常进度，warning 是快出问题的提醒，error 是已经出错。把这四类信息分清楚，排查项目问题会比满屏 print() 清楚很多。
核心源码（逐字来自文末完整源码）：
def demo_logging() -> None:
print("\n ── 日志系统演示 ──────────────────────────")
logger = setup_logger("geek.pipeline", level="DEBUG")
logger.debug("调试信息：变量 x=42")
logger.info("任务开始：数据采集")
logger.warning("内存使用率 85%，接近阈值")
logger.error("数据库连接失败，正在重试")
# 带上下文的日志（extra 字段）
extra_logger = logging.LoggerAdapter(logger, extra={"task_id": "T-001"})
# LoggerAdapter 不直接支持 extra 格式化，用 f-string 替代
logger.info("任务 T-001 完成，耗时 1.23s")
可运行演示（补齐 Mock 数据与 print 反馈）：
import logging
import sys
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""配置结构化日志器。"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
fmt=" %(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
# Step 4：同一个 logger 可以打 debug/info/warning/error，不同严重程度一眼可分。
def demo_logging() -> None:
print("\n ── 日志系统演示 ──────────────────────────")
logger = setup_logger("geek.pipeline", level="DEBUG")
logger.debug("调试信息：变量 x=42")
logger.info("任务开始：数据采集")
logger.warning("内存使用率 85%，接近阈值")
logger.error("数据库连接失败，正在重试")
# 带上下文的日志（extra 字段）
extra_logger = logging.LoggerAdapter(logger, extra={"task_id": "T-001"})
# LoggerAdapter 不直接支持 extra 格式化，用 f-string 替代
logger.info("任务 T-001 完成，耗时 1.23s")
demo_logging()
Step 5：用 pathlib + tempfile 安全练习目录和文件操作
痛点与机制：
pathlib 把路径当对象处理，base / "src" / "main.py" 比字符串拼接更不容易错。tempfile.TemporaryDirectory() 会创建临时目录，适合教学和测试：练完自动清理，不会把一堆练习文件留在桌面。
核心源码（逐字来自文末完整源码）：
def demo_pathlib() -> None:
print("\n ── pathlib 路径操作演示 ──────────────────")
# 用 tempfile 创建临时目录（零副作用）
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp)
# 创建目录结构
(base / "src").mkdir()
(base / "tests").mkdir()
(base / "data" / "raw").mkdir(parents=True)
# 写入文件
(base / "src" / "main.py").write_text('print("hello")\n')
(base / "src" / "utils.py").write_text("# utils\n")
(base / "README.md").write_text("# My Project\n")
(base / "data" / "raw" / "sample.csv").write_text("id,name\n1,Alice\n")
# 遍历目录树
print(f" 项目结构 ({base}):")
for path in sorted(base.rglob("*")):
depth = len(path.relative_to(base).parts) - 1
indent = " " + " " * depth
icon = "📁" if path.is_dir() else "📄"
size = f"({path.stat().st_size}B)" if path.is_file() else ""
print(f" {indent}{icon} {path.name} {size}")
# 路径操作
py_files = list(base.rglob("*.py"))
print(f"\n Python 文件数: {len(py_files)}")
for f in py_files:
print(f" {f.relative_to(base)} stem={f.stem} suffix={f.suffix}")
# 读取文件
content = (base / "data" / "raw" / "sample.csv").read_text()
print(f"\n CSV 内容:\n {content.strip()}")
可运行演示（补齐 Mock 数据与 print 反馈）：
import tempfile
from pathlib import Path
# Step 5：tempfile 像一次性练习场，脚本结束后不污染你的真实目录。
def demo_pathlib() -> None:
print("\n ── pathlib 路径操作演示 ──────────────────")
# 用 tempfile 创建临时目录（零副作用）
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp)
# 创建目录结构
(base / "src").mkdir()
(base / "tests").mkdir()
(base / "data" / "raw").mkdir(parents=True)
# 写入文件
(base / "src" / "main.py").write_text('print("hello")\n')
(base / "src" / "utils.py").write_text("# utils\n")
(base / "README.md").write_text("# My Project\n")
(base / "data" / "raw" / "sample.csv").write_text("id,name\n1,Alice\n")
# 遍历目录树
print(f" 项目结构 ({base}):")
for path in sorted(base.rglob("*")):
depth = len(path.relative_to(base).parts) - 1
indent = " " + " " * depth
icon = "📁" if path.is_dir() else "📄"
size = f"({path.stat().st_size}B)" if path.is_file() else ""
print(f" {indent}{icon} {path.name} {size}")
# 路径操作
py_files = list(base.rglob("*.py"))
print(f"\n Python 文件数: {len(py_files)}")
for f in py_files:
print(f" {f.relative_to(base)} stem={f.stem} suffix={f.suffix}")
# 读取文件
content = (base / "data" / "raw" / "sample.csv").read_text()
print(f"\n CSV 内容:\n {content.strip()}")
demo_pathlib()
Step 6：用 demo_json 处理 datetime 和 Path 这种复杂对象
痛点与机制：
JSON 只认识字符串、数字、列表、字典这些基础类型，datetime 和 Path 需要自定义 DateTimeEncoder 翻译一下。它像出国填表：你心里知道“今天下午三点”和“./data”是什么意思，但表格要求写成标准字符串。
核心源码（逐字来自文末完整源码）：
def demo_json() -> None:
print("\n ── JSON 序列化演示 ───────────────────────")
# 复杂对象序列化（自定义 encoder）
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj: Any) -> Any:
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Path):
return str(obj)
return super().default(obj)
data = {
"pipeline": "text-processing",
"created_at": datetime.now(),
"config": {"workers": 4, "data_dir": Path("./data")},
"tasks": [
{"id": 1, "name": "采集", "status": "done"},
{"id": 2, "name": "清洗", "status": "running"},
],
}
# 序列化
json_str = json.dumps(data, cls=DateTimeEncoder, ensure_ascii=False, indent=2)
print(f" 序列化结果（前5行）:")
for line in json_str.split("\n")[:5]:
print(f" {line}")
# 反序列化
loaded = json.loads(json_str)
print(f"\n 反序列化后 pipeline: {loaded['pipeline']}")
print(f" 任务数量: {len(loaded['tasks'])}")
# 用 tempfile 写入临时文件
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f, cls=DateTimeEncoder, ensure_ascii=False)
tmp_path = f.name
print(f"\n 已写入临时文件: {tmp_path}")
Path(tmp_path).unlink() # 清理
可运行演示（补齐 Mock 数据与 print 反馈）：
import json
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any
# Step 6：JSON 像跨系统通用表格，但 datetime/Path 需要先翻译成字符串。
def demo_json() -> None:
print("\n ── JSON 序列化演示 ───────────────────────")
# 复杂对象序列化（自定义 encoder）
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj: Any) -> Any:
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Path):
return str(obj)
return super().default(obj)
data = {
"pipeline": "text-processing",
"created_at": datetime.now(),
"config": {"workers": 4, "data_dir": Path("./data")},
"tasks": [
{"id": 1, "name": "采集", "status": "done"},
{"id": 2, "name": "清洗", "status": "running"},
],
}
# 序列化
json_str = json.dumps(data, cls=DateTimeEncoder, ensure_ascii=False, indent=2)
print(f" 序列化结果（前5行）:")
for line in json_str.split("\n")[:5]:
print(f" {line}")
# 反序列化
loaded = json.loads(json_str)
print(f"\n 反序列化后 pipeline: {loaded['pipeline']}")
print(f" 任务数量: {len(loaded['tasks'])}")
# 用 tempfile 写入临时文件
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f, cls=DateTimeEncoder, ensure_ascii=False)
tmp_path = f.name
print(f"\n 已写入临时文件: {tmp_path}")
Path(tmp_path).unlink() # 清理
demo_json()
Step 7：用 argparse 做总控开关，按 mode 调度不同模块
痛点与机制：
main() 把前面的独立小工具串起来。--mode config 只看配置，--mode path 只看路径，--mode all 全部跑一遍。命令行参数就是脚本遥控器，新手不用改源码也能切换功能。
核心源码（逐字来自文末完整源码）：
def main() -> None:
parser = argparse.ArgumentParser(description="模块化工具包演示")
parser.add_argument(
"--mode",
choices=["config", "logging", "path", "json", "all"],
default="all",
)
args = parser.parse_args()
if args.mode in ("config", "all"):
demo_config()
if args.mode in ("logging", "all"):
demo_logging()
if args.mode in ("path", "all"):
demo_pathlib()
if args.mode in ("json", "all"):
demo_json()
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import json
import logging
import os
import sys
import tempfile
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any
@dataclass
class AppConfig:
"""应用配置：优先读取环境变量，其次使用默认值。"""
app_name: str = field(default_factory=lambda: os.environ.get("APP_NAME", "geek-toolkit"))
log_level: str = field(default_factory=lambda: os.environ.get("LOG_LEVEL", "INFO"))
data_dir: Path = field(default_factory=lambda: Path(os.environ.get("DATA_DIR", "./data")))
max_workers: int = field(default_factory=lambda: int(os.environ.get("MAX_WORKERS", "4")))
debug: bool = field(default_factory=lambda: os.environ.get("DEBUG", "").lower() == "true")
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["data_dir"] = str(self.data_dir)
return d
def demo_config() -> None:
cfg = AppConfig()
print("\n ── 配置管理演示 ──────────────────────────")
print(f" {'配置项':<15} {'值'}")
print(f" {'─'*15} {'─'*25}")
for k, v in cfg.to_dict().items():
print(f" {k:<15} {v}")
# 环境变量覆盖演示
os.environ["APP_NAME"] = "my-custom-app"
cfg2 = AppConfig()
print(f"\n 设置 APP_NAME 环境变量后: {cfg2.app_name}")
del os.environ["APP_NAME"]
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""配置结构化日志器。"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
fmt=" %(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def demo_logging() -> None:
print("\n ── 日志系统演示 ──────────────────────────")
logger = setup_logger("geek.pipeline", level="DEBUG")
logger.debug("调试信息：变量 x=42")
logger.info("任务开始：数据采集")
logger.warning("内存使用率 85%，接近阈值")
logger.error("数据库连接失败，正在重试")
# 带上下文的日志（extra 字段）
extra_logger = logging.LoggerAdapter(logger, extra={"task_id": "T-001"})
# LoggerAdapter 不直接支持 extra 格式化，用 f-string 替代
logger.info("任务 T-001 完成，耗时 1.23s")
def demo_pathlib() -> None:
print("\n ── pathlib 路径操作演示 ──────────────────")
# 用 tempfile 创建临时目录（零副作用）
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp)
# 创建目录结构
(base / "src").mkdir()
(base / "tests").mkdir()
(base / "data" / "raw").mkdir(parents=True)
# 写入文件
(base / "src" / "main.py").write_text('print("hello")\n')
(base / "src" / "utils.py").write_text("# utils\n")
(base / "README.md").write_text("# My Project\n")
(base / "data" / "raw" / "sample.csv").write_text("id,name\n1,Alice\n")
# 遍历目录树
print(f" 项目结构 ({base}):")
for path in sorted(base.rglob("*")):
depth = len(path.relative_to(base).parts) - 1
indent = " " + " " * depth
icon = "📁" if path.is_dir() else "📄"
size = f"({path.stat().st_size}B)" if path.is_file() else ""
print(f" {indent}{icon} {path.name} {size}")
# 路径操作
py_files = list(base.rglob("*.py"))
print(f"\n Python 文件数: {len(py_files)}")
for f in py_files:
print(f" {f.relative_to(base)} stem={f.stem} suffix={f.suffix}")
# 读取文件
content = (base / "data" / "raw" / "sample.csv").read_text()
print(f"\n CSV 内容:\n {content.strip()}")
def demo_json() -> None:
print("\n ── JSON 序列化演示 ───────────────────────")
# 复杂对象序列化（自定义 encoder）
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj: Any) -> Any:
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Path):
return str(obj)
return super().default(obj)
data = {
"pipeline": "text-processing",
"created_at": datetime.now(),
"config": {"workers": 4, "data_dir": Path("./data")},
"tasks": [
{"id": 1, "name": "采集", "status": "done"},
{"id": 2, "name": "清洗", "status": "running"},
],
}
# 序列化
json_str = json.dumps(data, cls=DateTimeEncoder, ensure_ascii=False, indent=2)
print(f" 序列化结果（前5行）:")
for line in json_str.split("\n")[:5]:
print(f" {line}")
# 反序列化
loaded = json.loads(json_str)
print(f"\n 反序列化后 pipeline: {loaded['pipeline']}")
print(f" 任务数量: {len(loaded['tasks'])}")
# 用 tempfile 写入临时文件
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f, cls=DateTimeEncoder, ensure_ascii=False)
tmp_path = f.name
print(f"\n 已写入临时文件: {tmp_path}")
Path(tmp_path).unlink() # 清理
# Step 7：main 像工具箱总开关，--mode 决定今天演示哪个抽屉。
def main() -> None:
parser = argparse.ArgumentParser(description="模块化工具包演示")
parser.add_argument(
"--mode",
choices=["config", "logging", "path", "json", "all"],
default="all",
)
args = parser.parse_args()
if args.mode in ("config", "all"):
demo_config()
if args.mode in ("logging", "all"):
demo_logging()
if args.mode in ("path", "all"):
demo_pathlib()
if args.mode in ("json", "all"):
demo_json()
sys.argv = ["prog", "--mode", "config"]
main()
极客实战：完整源码与运行
现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。
# toolkit_demo.py
"""
模块化工具包演示 —— 标准库核心工具的工程级用法。
用法：
python3 toolkit_demo.py
python3 toolkit_demo.py --mode config
python3 toolkit_demo.py --mode logging
python3 toolkit_demo.py --mode path
python3 toolkit_demo.py --mode json
"""
import argparse
import json
import logging
import os
import sys
import tempfile
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any
# ══════════════════════════════════════════════════════════════
# 模拟 geek_toolkit/config.py
# ══════════════════════════════════════════════════════════════
@dataclass
class AppConfig:
"""应用配置：优先读取环境变量，其次使用默认值。"""
app_name: str = field(default_factory=lambda: os.environ.get("APP_NAME", "geek-toolkit"))
log_level: str = field(default_factory=lambda: os.environ.get("LOG_LEVEL", "INFO"))
data_dir: Path = field(default_factory=lambda: Path(os.environ.get("DATA_DIR", "./data")))
max_workers: int = field(default_factory=lambda: int(os.environ.get("MAX_WORKERS", "4")))
debug: bool = field(default_factory=lambda: os.environ.get("DEBUG", "").lower() == "true")
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["data_dir"] = str(self.data_dir)
return d
def demo_config() -> None:
cfg = AppConfig()
print("\n ── 配置管理演示 ──────────────────────────")
print(f" {'配置项':<15} {'值'}")
print(f" {'─'*15} {'─'*25}")
for k, v in cfg.to_dict().items():
print(f" {k:<15} {v}")
# 环境变量覆盖演示
os.environ["APP_NAME"] = "my-custom-app"
cfg2 = AppConfig()
print(f"\n 设置 APP_NAME 环境变量后: {cfg2.app_name}")
del os.environ["APP_NAME"]
# ══════════════════════════════════════════════════════════════
# 模拟 geek_toolkit/logger.py
# ══════════════════════════════════════════════════════════════
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""配置结构化日志器。"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
fmt=" %(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def demo_logging() -> None:
print("\n ── 日志系统演示 ──────────────────────────")
logger = setup_logger("geek.pipeline", level="DEBUG")
logger.debug("调试信息：变量 x=42")
logger.info("任务开始：数据采集")
logger.warning("内存使用率 85%，接近阈值")
logger.error("数据库连接失败，正在重试")
# 带上下文的日志（extra 字段）
extra_logger = logging.LoggerAdapter(logger, extra={"task_id": "T-001"})
# LoggerAdapter 不直接支持 extra 格式化，用 f-string 替代
logger.info("任务 T-001 完成，耗时 1.23s")
# ══════════════════════════════════════════════════════════════
# 模拟 geek_toolkit/path_utils.py
# ══════════════════════════════════════════════════════════════
def demo_pathlib() -> None:
print("\n ── pathlib 路径操作演示 ──────────────────")
# 用 tempfile 创建临时目录（零副作用）
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp)
# 创建目录结构
(base / "src").mkdir()
(base / "tests").mkdir()
(base / "data" / "raw").mkdir(parents=True)
# 写入文件
(base / "src" / "main.py").write_text('print("hello")\n')
(base / "src" / "utils.py").write_text("# utils\n")
(base / "README.md").write_text("# My Project\n")
(base / "data" / "raw" / "sample.csv").write_text("id,name\n1,Alice\n")
# 遍历目录树
print(f" 项目结构 ({base}):")
for path in sorted(base.rglob("*")):
depth = len(path.relative_to(base).parts) - 1
indent = " " + " " * depth
icon = "📁" if path.is_dir() else "📄"
size = f"({path.stat().st_size}B)" if path.is_file() else ""
print(f" {indent}{icon} {path.name} {size}")
# 路径操作
py_files = list(base.rglob("*.py"))
print(f"\n Python 文件数: {len(py_files)}")
for f in py_files:
print(f" {f.relative_to(base)} stem={f.stem} suffix={f.suffix}")
# 读取文件
content = (base / "data" / "raw" / "sample.csv").read_text()
print(f"\n CSV 内容:\n {content.strip()}")
# ══════════════════════════════════════════════════════════════
# 模拟 geek_toolkit/serializer.py
# ══════════════════════════════════════════════════════════════
def demo_json() -> None:
print("\n ── JSON 序列化演示 ───────────────────────")
# 复杂对象序列化（自定义 encoder）
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj: Any) -> Any:
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Path):
return str(obj)
return super().default(obj)
data = {
"pipeline": "text-processing",
"created_at": datetime.now(),
"config": {"workers": 4, "data_dir": Path("./data")},
"tasks": [
{"id": 1, "name": "采集", "status": "done"},
{"id": 2, "name": "清洗", "status": "running"},
],
}
# 序列化
json_str = json.dumps(data, cls=DateTimeEncoder, ensure_ascii=False, indent=2)
print(f" 序列化结果（前5行）:")
for line in json_str.split("\n")[:5]:
print(f" {line}")
# 反序列化
loaded = json.loads(json_str)
print(f"\n 反序列化后 pipeline: {loaded['pipeline']}")
print(f" 任务数量: {len(loaded['tasks'])}")
# 用 tempfile 写入临时文件
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f, cls=DateTimeEncoder, ensure_ascii=False)
tmp_path = f.name
print(f"\n 已写入临时文件: {tmp_path}")
Path(tmp_path).unlink() # 清理
def main() -> None:
parser = argparse.ArgumentParser(description="模块化工具包演示")
parser.add_argument(
"--mode",
choices=["config", "logging", "path", "json", "all"],
default="all",
)
args = parser.parse_args()
if args.mode in ("config", "all"):
demo_config()
if args.mode in ("logging", "all"):
demo_logging()
if args.mode in ("path", "all"):
demo_pathlib()
if args.mode in ("json", "all"):
demo_json()
if __name__ == "__main__":
main()
终端预期输出（--mode path）：
$ python3 toolkit_demo.py --mode path
── pathlib 路径操作演示 ──────────────────
项目结构 (/tmp/xxx):
📁 data
📁 raw
📄 sample.csv (14B)
📄 README.md (14B)
📁 src
📄 main.py (16B)
📄 utils.py (9B)
📁 tests
Python 文件数: 2
src/main.py stem=main suffix=.py
src/utils.py stem=utils suffix=.py
CSV 内容:
id,name
1,Alice
包结构最佳实践
# 包结构示例（需要创建 geek_toolkit/ 目录结构才能运行）
"""
# geek_toolkit/__init__.py
# 控制 from geek_toolkit import * 时暴露的接口
from .config import AppConfig
from .logger import setup_logger
__all__ = ["AppConfig", "setup_logger"]
__version__ = "0.1.0"
"""
# 包使用示例（需要先创建 geek_toolkit 包）
"""
# 使用包
from geek_toolkit import AppConfig, setup_logger
cfg = AppConfig()
logger = setup_logger("my_app", cfg.log_level)
"""
NexDo Time ⚡
5 分钟极客微操：把本篇的四个 demo_* 函数拆分到真实的文件中（config.py、logger.py、path_utils.py、serializer.py），创建 __init__.py 统一导出，然后在 main.py 中用 from geek_toolkit import ... 调用。
Don’t wait for next time, do it in the next moment.