如何将Airflow DAG生成的.py文件自动推送到GitHub Enterprise仓库？

1. 生成个人GitHub Enterprise Token #

• 登录你的GitHub Enterprise账号，进入「Settings → Developer settings → Personal access tokens → Generate new token」
• 勾选repo权限（私有仓库推送必备，公开仓库可选public_repo），按需添加过期时间，生成后务必保存好Token（仅显示一次）

2. 在Airflow中安全存储Token #

绝对不要硬编码Token，推荐两种安全存储方式：

• Airflow Variables：在Airflow UI的「Admin → Variables」中添加键值对，比如键为GHE_PERSONAL_TOKEN，值填你的Token
• Google Secrets Manager：将Token存入Secrets Manager，给运行DAG的谷歌服务账号配置该Secret的访问权限，后续在DAG中通过Google Cloud SDK读取

3. 编写推送任务代码 #

方式一：使用GitHub API（推荐，无需安装Git）

通过API直接上传/更新文件，适合轻量化自动化场景：

import requests
import base64
from airflow.models import Variable
from airflow.exceptions import AirflowException

def push_to_ghe(**context):
    # 从XCom获取生成的文件路径（假设生成文件的任务已将路径存入XCom）
    generated_file_path = context['ti'].xcom_pull(task_ids='generate_py_file')
    
    # 读取文件内容
    try:
        with open(generated_file_path, 'r', encoding='utf-8') as f:
            file_content = f.read()
    except Exception as e:
        raise AirflowException(f"读取生成文件失败: {str(e)}")
    
    # 配置GitHub Enterprise参数
    ghe_api_url = "https://你的GHE实例域名/api/v3"
    repo_owner = "你的组织名"
    repo_name = "目标仓库名"
    target_repo_path = "generated/auto_created_file.py"  # 文件在仓库中的存储路径
    target_branch = "main"  # 推送目标分支
    token = Variable.get("GHE_PERSONAL_TOKEN")
    headers = {"Authorization": f"token {token}"}

    # 检查文件是否已存在（用于更新场景）
    check_url = f"{ghe_api_url}/repos/{repo_owner}/{repo_name}/contents/{target_repo_path}?ref={target_branch}"
    check_response = requests.get(check_url, headers=headers)
    
    # 构造请求数据
    request_data = {
        "message": "Auto-generated file from Airflow DAG",
        "content": base64.b64encode(file_content.encode()).decode(),
        "branch": target_branch
    }
    
    if check_response.status_code == 200:
        # 文件已存在，添加SHA参数用于更新
        request_data["sha"] = check_response.json()["sha"]
        push_url = check_url
    elif check_response.status_code == 404:
        # 文件不存在，使用创建接口
        push_url = f"{ghe_api_url}/repos/{repo_owner}/{repo_name}/contents/{target_repo_path}"
    else:
        raise AirflowException(f"检查仓库文件状态失败: {check_response.text}")

    # 执行推送
    push_response = requests.put(push_url, json=request_data, headers=headers)
    if not push_response.ok:
        raise AirflowException(f"推送文件失败: {push_response.text}")

方式二：使用Git命令行（适合熟悉Git操作的场景）

需要确保Airflow运行环境已安装Git，代码示例：

import subprocess
from airflow.models import Variable
from airflow.exceptions import AirflowException

def push_to_ghe_via_git(**context):
    generated_file_path = context['ti'].xcom_pull(task_ids='generate_py_file')
    temp_repo_dir = "/tmp/temp_ghe_repo"
    ghe_repo_url = "https://你的GHE实例域名/你的组织名/目标仓库名.git"
    token = Variable.get("GHE_PERSONAL_TOKEN")
    # 构造带Token的授权URL，避免手动输入密码
    auth_repo_url = ghe_repo_url.replace("https://", f"https://{token}@")
    
    try:
        # 浅克隆目标分支（仅拉取最新代码，提升速度）
        subprocess.run(
            ["git", "clone", "--depth", "1", "--branch", "main", auth_repo_url, temp_repo_dir],
            check=True, capture_output=True, text=True
        )
        
        # 复制生成文件到仓库目标目录
        target_file_path = f"{temp_repo_dir}/generated/auto_created_file.py"
        subprocess.run(["cp", generated_file_path, target_file_path], check=True, capture_output=True, text=True)
        
        # 配置Git提交身份（必填）
        subprocess.run(["git", "-C", temp_repo_dir, "config", "user.name", "你的用户名"], check=True)
        subprocess.run(["git", "-C", temp_repo_dir, "config", "user.email", "你的邮箱"], check=True)
        
        # 提交并推送
        subprocess.run(["git", "-C", temp_repo_dir, "add", target_file_path], check=True)
        subprocess.run(
            ["git", "-C", temp_repo_dir, "commit", "-m", "Auto-generated file from Airflow DAG"],
            check=True, capture_output=True, text=True
        )
        subprocess.run(["git", "-C", temp_repo_dir, "push"], check=True, capture_output=True, text=True)
    except subprocess.CalledProcessError as e:
        raise AirflowException(f"Git操作失败: {e.stderr}")
    finally:
        # 清理临时仓库目录
        subprocess.run(["rm", "-rf", temp_repo_dir], capture_output=True)

4. 整合到Airflow DAG #

将推送任务添加到你的现有DAG中，示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'retries': 1,
    'retry_delay': 300
}

with DAG(
    'generate_and_push_to_ghe',
    default_args=default_args,
    schedule_interval=None,
    catchup=False
) as dag:
    # 你的生成文件任务
    generate_py_file = PythonOperator(
        task_id='generate_py_file',
        python_callable=你的生成文件函数,
        provide_context=True
    )

    # 推送任务（二选一）
    push_task = PythonOperator(
        task_id='push_to_ghe',
        python_callable=push_to_ghe,
        provide_context=True
    )

    generate_py_file >> push_task

注意事项 #

• Token权限最小化：只给repo权限即可，避免授予不必要的高权限
• 分支保护：如果目标分支有保护规则，确保你的个人账号有权限推送代码（比如添加到允许推送的用户列表）
• 错误处理：代码中已添加异常捕获，Airflow会根据retries配置自动重试失败任务

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