如何在任意Airflow Worker上获取mylib库的当前Git哈希值？

因为你的mylib是通过Git URL安装的，pip会自动记录相关版本元数据，以下几种通用方法可以帮你获取Git哈希，无需依赖Git环境或提前知道安装目录：

方法1：利用importlib.metadata（Python 3.8+） #

Python 3.8及以上自带的importlib.metadata模块可直接读取已安装包的元数据，从Git URL安装的包会包含哈希信息：

from importlib.metadata import version, PackageNotFoundError

def get_mylib_git_hash():
    try:
        # 通过Git URL安装时，version返回格式通常为 <版本号>+g<哈希值>，例如0.1.0+gabc1234
        full_version = version("mylib")
        if "+g" in full_version:
            return full_version.split("+g")[-1]
        return full_version
    except PackageNotFoundError:
        return "unknown"

在Airflow的PythonOperator中调用示例：

from airflow.operators.python import PythonOperator
import logging

def log_mylib_hash():
    git_hash = get_mylib_git_hash()
    logging.info(f"mylib 当前Git哈希值: {git_hash}")

task = PythonOperator(
    task_id="log_mylib_git_hash",
    python_callable=log_mylib_hash
)

方法2：读取pip生成的元数据文件（兼容Python 3.7及以下） #

如果你的Airflow环境使用旧版Python，可以手动读取pip安装时生成的元数据文件：

import os
import site
import json
from importlib.util import find_spec

def get_mylib_git_hash():
    # 定位mylib的安装路径
    spec = find_spec("mylib")
    if not spec or not spec.origin:
        return "unknown"
    
    # 查找dist-info目录
    package_name = "mylib"
    dist_info_dir = None
    for site_pkg_dir in site.getsitepackages():
        for dir_name in os.listdir(site_pkg_dir):
            if dir_name.startswith(f"{package_name}-") and dir_name.endswith(".dist-info"):
                dist_info_dir = os.path.join(site_pkg_dir, dir_name)
                break
        if dist_info_dir:
            break
    if not dist_info_dir:
        return "unknown"
    
    # 读取direct_url.json文件，其中包含Git提交信息
    direct_url_path = os.path.join(dist_info_dir, "direct_url.json")
    if os.path.exists(direct_url_path):
        with open(direct_url_path, "r") as f:
            data = json.load(f)
            if "vcs_info" in data:
                return data["vcs_info"]["commit_id"]
    return "unknown"

方法3：在mylib源码中内置哈希信息（最可靠） #

最稳定的方式是在mylib构建时自动嵌入Git哈希，通过setuptools-scm工具实现：

1. 在mylib的pyproject.toml中添加配置：

[build-system]
requires = ["setuptools>=61.0", "setuptools-scm"]
build-backend = "setuptools.build_meta"

[tool.setuptools_scm]

2. 在mylib的__init__.py中导入自动生成的版本信息：

from ._version import __version__, __git_hash__

之后在Airflow中直接读取即可：

import mylib
import logging

def log_mylib_hash():
    logging.info(f"mylib 当前Git哈希值: {mylib.__git_hash__}")

这种方法完全不依赖运行环境的Git或pip元数据，兼容性最强。

内容的提问来源于stack exchange，提问作者pyCthon