大数据

Python 工具库打包实战：从项目结构到 whl 包发布

Posted by

在团队协作或数据工程中，我们经常需要复用一些通用工具，例如重试机制、日志管理、文件扫描等。将这些功能封装成一个 Python 包，通过 pip install 安装，是最优雅的复用方式。本文将以 bigdata_utils 为例，手把手教你创建、打包并安装自己的 Python 工具库。

成品预览

最终我们将实现：

  • 清晰的模块分层：核心工具、文件IO、数据库、网络、浏览器自动化等各司其职
  • 统一导出：用户只需 from bigdata_utils import RetryMonitor 即可使用
  • 自动打包：运行 python -m build 生成 .whl 和源码包
  • 可编辑安装：开发时修改代码即时生效
  • 团队共享：whl 包可上传至内部 PyPI 或直接发给同事安装

1. 项目初始化与目录结构

我们采用 src 布局，这是目前 Python 打包的最佳实践，能自动避免误打包根目录下的 tests、.venv 等文件夹。

bigdata_utils/               # 项目根目录
├── src/
│   └── bigdata_utils/       # 真正的 Python 包
│       ├── __init__.py      # 顶层导出
│       ├── core/            # 核心基础工具
│       │   ├── __init__.py
│       │   ├── retry_monitor.py
│       │   └── logger_manager.py
│       ├── io/              # 文件与存储
│       │   ├── __init__.py
│       │   └── file_scanner.py
│       ├── db/              # 数据库工具
│       │   └── __init__.py
│       ├── net/             # 网络请求工具
│       │   └── __init__.py
│       └── browser/         # 浏览器自动化
│           └── __init__.py
├── .venv/                   # 虚拟环境（会被自动忽略）
├── pyproject.toml           # 打包配置核心
├── README.md
└── LICENSE

为什么用 src 布局？
setuptools 只会搜索 src 下的包，根目录的 tests、docs、.venv 等绝不会被打包，既安全又干净。

2. 编写工具类（核心示例）

2.1 重试监控器（core/retry_monitor.py）

import logging
import traceback
from datetime import datetime
from tenacity import RetryCallState

class RetryMonitor:
    """tenacity 重试回调监控，输出标准化日志"""

    def __init__(self, logger: logging.Logger = None, level: int = logging.WARNING):
        self.logger = logger or logging.getLogger(__name__)
        self.level = level

    @staticmethod
    def _format_time() -> str:
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    @staticmethod
    def _get_error_detail(exception: Exception):
        error_type = type(exception).__name__
        error_msg = str(exception)
        tb_lines = traceback.format_exc().split("\n")
        error_location = "未知位置"
        for line in reversed(tb_lines):
            if line.strip().startswith("File"):
                error_location = line.strip()
                break
        return error_type, error_msg, error_location

    def before_sleep_callback(self, retry_state: RetryCallState):
        exception = retry_state.outcome.exception()
        error_type, error_msg, error_location = self._get_error_detail(exception)
        wait_seconds = retry_state.next_action.sleep if retry_state.next_action else 0
        msg = (
            f"\n{'=' * 60}\n"
            f"🔄 准备第 {retry_state.attempt_number} 次重试\n"
            f"⏰ 时间: {self._format_time()}\n"
            f"💥 错误类型: {error_type}\n"
            f"📝 错误信息: {error_msg}\n"
            f"📍 错误位置: {error_location}\n"
            f"⏳ 等待 {wait_seconds:.1f} 秒后重试...\n"
            f"{'=' * 60}"
        )
        self.logger.log(self.level, msg)

    def after_attempt_callback(self, retry_state: RetryCallState):
        exception = retry_state.outcome.exception()
        error_type, error_msg, _ = self._get_error_detail(exception)
        msg = f"❌ 第 {retry_state.attempt_number} 次执行失败 -> {error_type}: {error_msg}"
        self.logger.log(self.level, msg)

2.2 日志管理器（core/logger_manager.py）

import logging
import os
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

class LoggerManager:
    _initialized = False

    @classmethod
    def init(cls, log_dir="logs", log_file_name="app.log",
             file_log_level=logging.DEBUG, console_log_level=logging.INFO,
             file_rotation="size", max_bytes=10*1024*1024, backup_count=5,
             when="midnight", console_only=False):
        if cls._initialized:
            return
        root = logging.getLogger()
        root.setLevel(logging.DEBUG)
        root.handlers.clear()
        fmt = logging.Formatter(
            "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        # 控制台 handler
        ch = logging.StreamHandler()
        ch.setLevel(console_log_level)
        ch.setFormatter(fmt)
        root.addHandler(ch)

        if not console_only:
            os.makedirs(log_dir, exist_ok=True)
            log_path = os.path.join(log_dir, log_file_name)
            if file_rotation == "size":
                fh = RotatingFileHandler(log_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8")
            elif file_rotation == "time":
                fh = TimedRotatingFileHandler(log_path, when=when, backupCount=backup_count, encoding="utf-8")
            else:
                fh = logging.FileHandler(log_path, encoding="utf-8")
            fh.setLevel(file_log_level)
            fh.setFormatter(fmt)
            root.addHandler(fh)
        cls._initialized = True

    @classmethod
    def get_logger(cls, name=None):
        if not cls._initialized:
            raise RuntimeError("请先调用 LoggerManager.init() 初始化日志")
        return logging.getLogger(name)

2.3 文件扫描器（io/file_scanner.py）

from pathlib import Path
from typing import List, Optional

class FileScanner:
    @staticmethod
    def scan(directory: str, extensions: Optional[List[str]] = None, recursive: bool = True) -> List[Path]:
        dir_path = Path(directory)
        if not dir_path.is_dir():
            raise NotADirectoryError(f"{directory} 不是有效目录")
        pattern = "**/*" if recursive else "*"
        files = list(dir_path.glob(pattern))
        if extensions:
            exts = [ext.lower() if ext.startswith('.') else f'.{ext.lower()}' for ext in extensions]
            files = [f for f in files if f.is_file() and f.suffix.lower() in exts]
        else:
            files = [f for f in files if f.is_file()]
        return sorted(files)

2.4 统一导出（各层 __init__.py）

core/__init__.py

from .retry_monitor import RetryMonitor
from .logger_manager import LoggerManager

io/__init__.py

from .file_scanner import FileScanner

顶层 bigdata_utils/__init__.py

from .core import RetryMonitor, LoggerManager
from .io import FileScanner

__version__ = "0.1.0"

现在，安装后用户可以直接：

from bigdata_utils import RetryMonitor, LoggerManager, FileScanner

3. 配置 pyproject.toml

这是打包的核心配置文件，放在项目根目录。

[build-system]
requires = ["setuptools>=64", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "bigdata_utils"
version = "0.1.0"
description = "大数据开发与治理内部工具集（重试、日志、文件扫描等）"
authors = [{name = "Your Team", email = "team@example.com"}]
readme = "README.md"
license = "MIT"
license-files = ["LICENSE"]
requires-python = ">=3.8"
dependencies = [
    "tenacity>=8.0",
]

[tool.setuptools]
package-dir = {"" = "src"}

[tool.setuptools.packages.find]
where = ["src"]

关键说明：

  • license 使用 SPDX 标识（如 MIT），避免弃用警告
  • license-files 指定要包含的许可证文件
  • dependencies 列出运行时依赖（如 tenacity）
  • [tool.setuptools] 指明包在 src 目录下

4. 构建 whl 包

确保已安装构建工具：

pip install build

在项目根目录运行：

python -m build

输出会显示

Successfully built bigdata_utils-0.1.0.tar.gz and bigdata_utils-0.1.0-py3-none-any.whl

生成的文件在 dist/ 目录中。

验证 whl 内容（可选）：

unzip -l dist/bigdata_utils-0.1.0-py3-none-any.whl

5. 安装与使用

5.1 直接安装 whl

pip install dist/bigdata_utils-0.1.0-py3-none-any.whl

5.2 开发模式安装（推荐开发时使用）

在项目根目录执行：

pip install -e .

这会以“可编辑”模式安装，源码的修改会立即生效，无需重新打包。

5.3 在项目中使用

from bigdata_utils import LoggerManager, RetryMonitor, FileScanner
import logging
from tenacity import retry, stop_after_attempt, before_sleep, after

# 初始化日志
LoggerManager.init(log_dir="logs", log_file_name="pipeline.log")

# 扫描文件
files = FileScanner.scan("data", extensions=[".csv", ".txt"])

# 重试监控
monitor = RetryMonitor(level=logging.WARNING)

@retry(stop=stop_after_attempt(3),
       before_sleep=monitor.before_sleep_callback,
       after=monitor.after_attempt_callback)
def process_file(file_path):
    ...

6. 进阶：分子包与内部调用

当工具增多时，按功能分 core、io、db、net、browser 等子包。
内部调用直接使用绝对导入，例如 net 模块需要 RetryMonitor

from bigdata_utils.core.retry_monitor import RetryMonitor

只要包已安装，这种导入就能正常解析。

避免循环导入原则：

  • core 作为底层不依赖任何兄弟包
  • io、db、net 可以依赖 core
  • 如果两个子包需要互相调用，考虑将共享逻辑下沉到 core，或使用延迟导入（函数内 import）

7. 常见问题与技巧

7.1 如何排除 .venv 等文件夹？

使用 src 布局后，setuptools 只搜索 src 下的包，根目录的 .venv、tests 等不会被打包。如果采用扁平布局，可在 pyproject.toml 中添加：

[tool.setuptools.packages.find]
exclude = [".venv*"]

7.2 构建时出现 license 字段弃用警告？

将 license = {file = "LICENSE"} 改为：

license = "MIT"
license-files = ["LICENSE"]

7.3 如何控制日志文件的滚动？

LoggerManager.init() 支持：

  • file_rotation="size"：按大小滚动，配合 max_bytes 和 backup_count
  • file_rotation="time"：按时间滚动，配合 when（如 "midnight"）

7.4 如何在团队间分享？

  • 直接发送 .whl 文件，对方 pip install xxx.whl
  • 搭建私有 PyPI（如 pypiserver 或 GitLab PyPI），配置后可直接 pip install bigdata_utils

上一篇 tenacity 重试监控实战